内容简介:用户有个内部系统,提供基础数据,用户想在不改造改内部系统的前提下将数据文件(txt)内容进行加密,然后copy到我们系统思来想去只能搞一个代理服务器拦截用户的下载请求,解析内容后返回加密后文件
用户有个内部系统,提供基础数据,用户想在不改造改内部系统的前提下将数据文件(txt)内容进行加密,然后copy到我们系统
思路
思来想去只能搞一个代理服务器拦截用户的下载请求,解析内容后返回加密后文件
参考
具体实现代码
import logging; logging.basicConfig(level=logging.INFO) import socket, select import _thread from io import BytesIO from Cryptodome.Cipher import AES class Proxy(object): def __init__(self, soc): self.client, _ = soc.accept() self.target = None self.is_export = False self.BUFSIZE = 1024 self.method = None self.targetHost = None self.s = None # 解析请求 def getClientRequest(self): request = self.client.recv(self.BUFSIZE).decode() if not request: return None cn = request.find('\n') firstLine = request[:cn] line = firstLine.split() self.method = line[0] self.targetHost = line[1] return request # 拦截正常请求 def commonMethod(self, request): tmp = self.targetHost.split('/') logging.info(tmp) targetAddr = self.getTargetInfo(tmp[2]) if len(tmp) > 5 and tmp[5].find('export_delivery') >= 0: self.is_export = True else: self.is_export = False try: (fam, _, _, _, addr) = socket.getaddrinfo(targetAddr[0], targetAddr[1])[0] except Exception as e: print(e) return self.target = socket.socket(fam) self.target.connect(addr) self.target.send(request.encode()) self.nonblocking() def connectMethod(self, request): print('建立连接') pass # 启动方法 def run(self): request = self.getClientRequest() if request: if self.method in ['GET', 'POST', 'PUT', 'DELETE', 'HAVE']: self.commonMethod(request) elif self.method == 'CONNECT': self.connectMethod(request) # 分析数据 def nonblocking(self): inputs = [self.client, self.target] break_flag = False if (self.is_export == True): self.s = BytesIO() while True: if break_flag == True: break readable, writeable, errs = select.select(inputs, [], inputs, 3) if errs: print('nonblocking errs') break for soc in readable: data = soc.recv(self.BUFSIZE) if data: if soc is self.client: self.target.send(data) elif soc is self.target: if (self.is_export == True): self.s.write(data) else: self.client.send(data) else: break_flag = True break if (self.is_export == True): self.parseRequest() self.client.close() self.target.close() # 解析文件内容并加密 def parseRequest(self): try: _res = self.s.getvalue().decode("gb2312") tmp = _res.split('octet-stream\r\n\r\n') _h = tmp[0] + 'octet-stream\r\n\r\n' _b = tmp[1] nb = BytesIO() nb.write(_h.encode('utf8')) secret_key = "ThisIs SecretKey" iv_param = 'This is an IV456' aes1 = AES.new(secret_key.encode("gb2312"), AES.MODE_CFB, iv_param.encode("gb2312")) cipher_data = aes1.encrypt(_b.encode("gb2312")) nb.write(cipher_data) self.client.send(nb.getvalue()) except Exception as e: print('Error:', e) self.client.send(self.s.getvalue()) def getTargetInfo(self, host): port = 0 site = None if ':' in host: tmp = host.split(':') site = tmp[0] port = int(tmp[1]) else: site = host port = 80 return site, port if __name__ == '__main__': host = '127.0.0.1' port = 8083 backlog = 5 server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server.bind((host, port)) server.listen(backlog) while True: # 多线程 # t = Process(target=Proxy(server).run) # t.start() # 单线程 _thread.start_new_thread(Proxy(server).run, ()) 复制代码
以上所述就是小编给大家介绍的《Python 之构建代理服务器》,希望对大家有所帮助,如果大家有任何疑问请给我留言,小编会及时回复大家的。在此也非常感谢大家对 码农网 的支持!
猜你喜欢:- 《高性能linux服务器构建实战》
- 使用Freya构建F# Web服务器
- 使用Gulp构建本地开发Web服务器
- 重新思考使用无服务器构建软件的方式
- 使用 Solr 构建企业级搜索服务器
- 利用LXD容器构建共享的GPU服务器
本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们。
用户至上:用户研究方法与实践(原书第2版)
凯茜·巴克斯特 / 王兰、杨雪、苏寅 / 机械工业出版社 / 2017-5-1 / 99
《UI/UE系列丛书 用户至上:用户研究方法与实践(原书第2版)》是用户研究方法指南,谷歌用户体验研究员十几年工作经验结晶,从理论到实战,包含完整的实操案例,是设计以人为中心产品的实用手册。 《UI/UE系列丛书 用户至上:用户研究方法与实践(原书第2版)》包含五个部分共15章。入门篇包括第1~5章:介绍用户体验入门,如何理解目标用户,道德与法律问题,如何搭建研究设施,如何选择用户体验研究方......一起来看看 《用户至上:用户研究方法与实践(原书第2版)》 这本书的介绍吧!