python之路----验证客户端合法性

时间：2023-03-08 19:23:00

验证客户端链接的合法性

import os

import hmac

import socket

secret_key = b'egg'

sk = socket.socket()

sk.bind(('127.0.0.1',8080))

sk.listen()

def check_conn(conn):

    msg = os.urandom(32)

    conn.send(msg)

    h = hmac.new(secret_key,msg)

    digest = h.digest()

    client_digest = conn.recv(1024)

    return hmac.compare_digest(digest,client_digest)

conn,addr = sk.accept()

res = check_conn(conn)

if res:

    print('合法的客户端')

    conn.close()

else:

    print('不合法的客户端')

    conn.close()

sk.close()

server端

import hmac

import socket

secret_key = b'egg'

sk = socket.socket()

sk.connect(('127.0.0.1',8080))

msg = sk.recv(1024)

h = hmac.new(secret_key,msg)

digest = h.digest()

sk.send(digest)

sk.close()

client端

利用hmac+加盐的方式在分布式系统中实现一个简单的客户端链接认证功能

#_*_coding:utf-8_*_

from socket import *

import hmac,os

secret_key=b'linhaifeng bang bang bang'

def conn_auth(conn):

    '''

    认证客户端链接

    :param conn:

    :return:

    '''

    print('开始验证新链接的合法性')

    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('该链接不合法,关闭')

        conn.close()

        return

    print('链接合法,开始通信')

    while True:

        data=conn.recv(bufsize)

        if not data:break

        conn.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=1024

    server_handler(ip_port,bufsize)

server端

#_*_coding:utf-8_*_

__author__ = 'Linhaifeng'

from socket import *

import hmac,os

secret_key=b'linhaifeng bang bang bang'

def conn_auth(conn):

    '''

    验证客户端到服务器的链接

    :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:continue

        if data == 'quit':break

        tcp_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=1024

    client_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:continue

        if data == 'quit':break

        tcp_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=1024

    client_handler(ip_port,bufsize)

客户端(非法:不知道加密方式)

#_*_coding:utf-8_*_

__author__ = 'Linhaifeng'

from socket import *

import hmac,os

secret_key=b'linhaifeng bang bang bang1111'

def conn_auth(conn):

    '''

    验证客户端到服务器的链接

    :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:continue

        if data == 'quit':break

        tcp_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=1024

    client_handler(ip_port,bufsize)

客户端(非法:不知道secret_key)

相关文章