python ssl wrap

通过python可以免去编译很快速的实现ssl socket的模拟，如下server和client的示例，作为日常学习ssl的行为模式，应该是能满足了： sslServer:

import socket

import ssl

import sys

def sslServer(host, port):

# create user defined SSLContext

ssl_context = ssl.SSLContext()

ssl_context.verify_mode = ssl.CERT_REQUIRED

ssl_context.load_verify_locations(cafile='RootCA.pem')

ssl_context.check_hostname = False

# test this protocol with client's PROTOCOL_TLSv1_2

ssl_context.protocol = ssl.PROTOCOL_TLS_SERVER

ssl_context.load_cert_chain("Server.pem", "Server.key")

cipher_suites = 'ECDHE-RSA-AES256-GCM-SHA384:AES256-GCM-SHA384:DHE-RSA-AES256-GCM-SHA384:ECDHE-RSA-AES256-GCM-SHA384'

ssl_context.set_ciphers(cipher_suites)

try:

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)

# set reuse options

server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

server.bind((host, port))

server.listen(1)

with ssl_context.wrap_socket(server, server_side=True) as ssl_server:

conn, addr = ssl_server.accept()

print("accept conn from %s" % str(addr))

# get the ssl cipher info

print('ssl cipher: {}'.format(conn.cipher()))

while conn:

recv_data = conn.recv(1024)

print("recv: %s" % repr(recv_data))

if not recv_data:

break

conn.sendall(recv_data)

except Exception as e:

print(e)

if __name__ == '__main__':

print('ssl ver: {}'.format(ssl.OPENSSL_VERSION))

try:

sslServer(sys.argv[1], int(sys.argv[2]))

except KeyboardInterrupt:

print('server exit')

sslClient:

import socket

import ssl

import sys

def sslClient(host, port):

# create defauly_context

ssl_context = ssl.create_default_context(cafile='RootCA.pem')

ssl_context.load_cert_chain("Client.pem", "Client.key")

ssl_context.protocol = ssl.PROTOCOL_TLSv1_2

# crl test

# ssl_context.verify_flags = ssl.VERIFY_CRL_CHECK_LEAF

# ssl_context.load_verify_locations("Server.crl")

cipher_suites = 'DHE-RSA-AES256-GCM-SHA384:ECDHE-RSA-AES256-GCM-SHA384'

ssl_context.set_ciphers(cipher_suites)

client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

with ssl_context.wrap_socket(client, server_hostname='Server') as ssl_client:

ssl_client.connect((host, port))

# get the ssl cipher info

print('ssl cipher: {}'.format(ssl_client.cipher()))

while True:

send_data = input("send:")

ssl_client.sendall(send_data.encode())

recv_data = ssl_client.recv(1024)

print("recv:%s" % recv_data)

if __name__ == '__main__':

print('ssl ver: {}'.format(ssl.OPENSSL_VERSION))

try:

sslClient(sys.argv[1], int(sys.argv[2]))

except KeyboardInterrupt:

print('client exit')

参考：官方文档: https://docs.python.org/3 […]

BaseHTTPServer

测试代码，留备参考。python3相对pyhton2的改变还是不少，比如发送/接收之前之后的编码，包的导入方式等

# -*- coding: utf-8 -*-

#python3.6

from http.server import BaseHTTPRequestHandler, HTTPServer

import json

import socketserver

import threading

class myHTTPHandler(BaseHTTPRequestHandler):

def _set_header(self):

self.protocol = 'HTTP/1.1'

self.send_response(200, 'OK')

self.send_header('Content-type', 'application/json')

self.end_headers()

def do_GET(self):

#获取线程名称，对比myHTTPServer和HTTPServer的线程差异

print(threading.current_thread().getName())

# 获取请求类型和地址（可以通过不抑制log_message来得到，一样的效果，此处只是为了留存）

print(self.command, self.path, self.client_address)

#显示header内容

print(self.headers)

#获取header指定属性的值

print('Accept-Language', self.headers.get('Accept-Language'))

#读取请求内容

#如果有Content-Length，可以通过长度来读取发送的数据内容

length = self.headers.get('Content-Length')

if length is not None:

rdata = self.rfile.read(int(length))

print(rdata.decode())

jsobBody = json.dumps(rdata.decode())

# 发送响应

self._set_header()

# pyhton3必须添加encode()才能正常发送

self.wfile.write(json.dumps({'ret1': 'hello world'}).encode())

def do_POST(self):

pass

# log处理函数，通过这种方式可以抑制请求消息输出

# def log_message(self, format, *args):

# pass

class myHTTPServer(socketserver.ThreadingMixIn, HTTPServer):

allow_reuse_address = True

pass

if __name__ == '__main__':

server = myHTTPServer(('127.0.0.1', 9999), myHTTPHandler)

try:

server.serve_forever()

except KeyboardInterrupt:

pass

server.server_close()

通过下面的curl命令可以测试带数据的POST或者GET，否则可以直接用网页来访问测试：

curl -H "Content-Type:application/json" -X GET --data '{"test": "hello from client"}' http://127.0.0.1:9999

关于putty的一些事

putty，command方式操作的工具是plink putty，自带sftp工具叫做psftp putty，自带scp工具叫做pscp 最后，最主要的，putty当前的版本是0.70，你可以下载putty.zip这个文件，里面包含putty的官方文档。

re子组和xml.etree.ElementTree写文件

关键信息临时中转╭(╯^╰)╮，

import xml.etree.ElementTree as ET

import re

def handle_version_str(src_str):

# 获取匹配的自组数据

m = re.search(r'(.*)(V\d{3}R\d{3}C\d{2})(.*)', src_str)

if m is not None:

ver = m.group(2)

print(ver)

# 通过获取的匹配内容，将其删除

dst_str = re.sub(ver, '', src_str)

else:

return None

return dst_str

def create_result_xml():

model = '''<data><country name="China">test</country></data>'''

root = ET.fromstring(model)

# 添加子节点

sub_attrib = {"location": "cn", "color":"black"}

root.append("sub1", sub_attrib)

# 如何通过ET写入不是parse方式创建的ET到一个xml文件

tree = ET.ElementTree(root)

tree.write('result.xml')

if __name__ == "__main__":

ver_str = "Hello_V001R002C20_Release"

dst_str = handle_version_str(ver_str)

print(dst_str)

create_result_xml()

socketserver

#-*- coding: utf-8 -*-

import socketserver

import socket

"""

测试环境：Ubuntu

Pyhton3.6.5

web访问: http://127.0.0.1:9999 通过

curl: curl http://127.0.0.1:9999 通过

"""

class myHandler(socketserver.StreamRequestHandler):

def handle(self):

self.rdata = self.rfile.readline()

# 测试for方法独到最后会持续阻塞

# for item in self.rfile:

# print("{} write: {}".format(self.client_address[0], item))

self.function1()

self.wfile.write(b'HTTP/1.1 200 OK\r\nContent-Type: text/html\r\n\r\n<html>hello</html>')

def function1(self):

if 'GET' in self.rdata.decode('utf-8'):

self.function2()

def function2(self):

"""测试rfile中的数据可以缓存后续继续read"""

while True:

item = self.rfile.readline()

# 此处无法设置非阻塞，通过判断http请求中最后一组数据为\r\n来退出循环

if item == b"\r\n":

break

print("{} write: {}".format(self.client_address[0], item))

if __name__ == "__main__":

# 全局设置端口重用

socketserver.ThreadingTCPServer.allow_reuse_address = True

server = socketserver.ThreadingTCPServer(('127.0.0.1', 9999), myHandler)

# 设置端口重用

# server.allow_reuse_address = True

# 网上关于设置socket参数方法

# server.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)

# 设置非阻塞不生效

server.socket.setblocking(False)

server.serve_forever()

python2: SockServer python3: socketserver 如果pyhton2一直提示找不到模块，尝试 pip install Werkzeug --upgrade

一	二	三	四	五	六	日
					1	2
3	4	5	6	7	8	9
10	11	12	13	14	15	16
17	18	19	20	21	22	23
24	25	26	27	28	29	30
31

mowblog

一群散乱的记忆

月度归档： 2018 年 12 月

python ssl wrap

BaseHTTPServer

关于putty的一些事

re子组和xml.etree.ElementTree写文件

socketserver