驗(yàn)證客戶端鏈接的合法性

分布式系統(tǒng)中實(shí)現(xiàn)一個簡單的客戶端鏈接認(rèn)證功能

#_*_coding:utf-8_*_ from socket import * import hmac,ossecret_key=b'linhaifeng bang bang bang' def conn_auth(conn):'''認(rèn)證客戶端鏈接:param conn::return:'''print('開始驗(yàn)證新鏈接的合法性')msg=os.urandom(32)conn.sendall(msg)h=hmac.new(secret_key,msg)digest=h.digest()respone=conn.recv(len(digest))return hmac.compare_digest(respone,digest)def data_handler(conn,bufsize=1024):if not conn_auth(conn):print('該鏈接不合法,關(guān)閉')conn.close()returnprint('鏈接合法,開始通信')while True:data=conn.recv(bufsize)if not data:breakconn.sendall(data.upper())def server_handler(ip_port,bufsize,backlog=5):'''只處理鏈接:param ip_port::return:'''tcp_socket_server=socket(AF_INET,SOCK_STREAM)tcp_socket_server.bind(ip_port)tcp_socket_server.listen(backlog)while True:conn,addr=tcp_socket_server.accept()print('新連接[%s:%s]' %(addr[0],addr[1]))data_handler(conn,bufsize)if __name__ == '__main__':ip_port=('127.0.0.1',9999)bufsize=1024server_handler(ip_port,bufsize)服務(wù)端 服務(wù)端 #_*_coding:utf-8_*_ __author__ = 'Linhaifeng' from socket import * import hmac,ossecret_key=b'linhaifeng bang bang bang' def conn_auth(conn):'''驗(yàn)證客戶端到服務(wù)器的鏈接:param conn::return:'''msg=conn.recv(32)h=hmac.new(secret_key,msg)digest=h.digest()conn.sendall(digest)def client_handler(ip_port,bufsize=1024):tcp_socket_client=socket(AF_INET,SOCK_STREAM)tcp_socket_client.connect(ip_port)conn_auth(tcp_socket_client)while True:data=input('>>: ').strip()if not data:continueif data == 'quit':breaktcp_socket_client.sendall(data.encode('utf-8'))respone=tcp_socket_client.recv(bufsize)print(respone.decode('utf-8'))tcp_socket_client.close()if __name__ == '__main__':ip_port=('127.0.0.1',9999)bufsize=1024client_handler(ip_port,bufsize)客戶端(合法) 客戶端 #_*_coding:utf-8_*_ __author__ = 'Linhaifeng' from socket import *def client_handler(ip_port,bufsize=1024):tcp_socket_client=socket(AF_INET,SOCK_STREAM)tcp_socket_client.connect(ip_port)while True:data=input('>>: ').strip()if not data:continueif data == 'quit':breaktcp_socket_client.sendall(data.encode('utf-8'))respone=tcp_socket_client.recv(bufsize)print(respone.decode('utf-8'))tcp_socket_client.close()if __name__ == '__main__':ip_port=('127.0.0.1',9999)bufsize=1024client_handler(ip_port,bufsize)客戶端(非法:不知道加密方式) 客戶端(非法:不知道加密方式) #_*_coding:utf-8_*_ __author__ = 'Linhaifeng' from socket import * import hmac,ossecret_key=b'linhaifeng bang bang bang1111' def conn_auth(conn):'''驗(yàn)證客戶端到服務(wù)器的鏈接:param conn::return:'''msg=conn.recv(32)h=hmac.new(secret_key,msg)digest=h.digest()conn.sendall(digest)def client_handler(ip_port,bufsize=1024):tcp_socket_client=socket(AF_INET,SOCK_STREAM)tcp_socket_client.connect(ip_port)conn_auth(tcp_socket_client)while True:data=input('>>: ').strip()if not data:continueif data == 'quit':breaktcp_socket_client.sendall(data.encode('utf-8'))respone=tcp_socket_client.recv(bufsize)print(respone.decode('utf-8'))tcp_socket_client.close()if __name__ == '__main__':ip_port=('127.0.0.1',9999)bufsize=1024client_handler(ip_port,bufsize)客戶端(非法:不知道secret_key) 客戶端(非法:不知道secret_key)

socketserver 模塊

實(shí)現(xiàn)基于TCP協(xié)議下的服務(wù)端與客戶端的一對多的交互模式

import socketserver class MyServer(socketserver.BaseRequestHandler):def handle(self): # self.request 就相當(dāng)于一個connwhile True:print(self.client_address)msg = self.request.recv(1024).decode('utf-8')if msg == 'q':breakprint(msg)info = input('%s>>>'%msg[:2])self.request.send(info.encode('utf-8'))if __name__ == '__main__':server = socketserver.ThreadingTCPServer(('127.0.0.1',8080),MyServer)# thread 線程server.allow_reuse_address = Trueserver.serve_forever() 服務(wù)端 import socket sk = socket.socket() sk.connect(('127.0.0.1',8080)) while True:msg = input('>>>')if msg == 'q':sk.send(b'q')breaksk.send(('美團(tuán) :'+msg).encode('utf-8'))ret = sk.recv(1024).decode('utf-8')print(ret) sk.close() 客戶端

?

轉(zhuǎn)載于:https://www.cnblogs.com/DI-DIAO/p/8386892.html

總結(jié)

以上是生活随笔為你收集整理的Python之socketserver模块和验证客户端链接的合法性的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。