Skip to content

Instantly share code, notes, and snippets.

@oborichkin
Last active November 10, 2024 04:14
Show Gist options
  • Save oborichkin/d8d0c7823fd6db3abeb25f69352a5299 to your computer and use it in GitHub Desktop.
Save oborichkin/d8d0c7823fd6db3abeb25f69352a5299 to your computer and use it in GitHub Desktop.
Simple TLS client and server on python
import socket
import ssl
from tls_server import HOST as SERVER_HOST
from tls_server import PORT as SERVER_PORT
HOST = "127.0.0.1"
PORT = 60002
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
client = ssl.wrap_socket(client, keyfile="path/to/keyfile", certfile="path/to/certfile")
if __name__ == "__main__":
client.bind((HOST, PORT))
client.connect((SERVER_HOST, SERVER_PORT))
while True:
from time import sleep
client.send("Hello World!".encode("utf-8"))
sleep(1)
import socket
import ssl
HOST = "127.0.0.1"
PORT = 60000
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server = ssl.wrap_socket(
server, server_side=True, keyfile="path/to/keyfile", certfile="path/to/certfile"
)
if __name__ == "__main__":
server.bind((HOST, PORT))
server.listen(0)
while True:
connection, client_address = server.accept()
while True:
data = connection.recv(1024)
if not data:
break
print(f"Received: {data.decode('utf-8')}")
@anson2416
Copy link

Hey @oborichkin , may I know how I could create a thread to handle client connection?
Giving the code, if a client disconnected, the server side will exit. I do not want the server side pgm ended abnormally.

@nns33213
Copy link

nns33213 commented May 5, 2023

Quite concerning that this gist is so popular for some reason. It's remarkably bad.


from tls_server import HOST as SERVER_HOST
from tls_server import PORT as SERVER_PORT

Just why...


PORT = 60002

Ports 49152–65535 are ephemeral ports. It's not recommended to use them.


client.bind((HOST, PORT))

You do not need to bind in client. OS will automatically pick an ephemeral source port.


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

Ok, I guess you found a fix for the useless thing above.


    while True:
        from time import sleep

Move that to the top. It's also not a good idea to import directly to a global namespace.


client.send("Hello World!".encode("utf-8"))

Use .sendall instead of .send in synchronous sockets to avoid surprises.

Fixed

import socket
import ssl
import time

SERVER_HOST = "127.0.0.1"
SERVER_PORT = 40000

if __name__ == "__main__":
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    client = ssl.wrap_socket(client, keyfile="path/to/keyfile", certfile="path/to/certfile")

    client.connect((SERVER_HOST, SERVER_PORT))

    while True:
        client.sendall("Hello World!".encode("utf-8"))

        time.sleep(1)
import socket
import ssl

HOST = "127.0.0.1"
PORT = 40000

if __name__ == "__main__":
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

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

    server = ssl.wrap_socket(
        server, server_side=True, keyfile="path/to/keyfile", certfile="path/to/certfile"
    )

    server.bind((HOST, PORT))
    server.listen(0)

    while True:
        connection, client_address = server.accept()

        while True:
            data = connection.recv(1024)

            if not data:
                break

            print(f"Received: {data.decode('utf-8')}")

@Slayerx96
Copy link

To avoid getting the deprecation warning and update the code SSLContext method must be used and the unwrapped socket closed as specified at https://pythontic.com/ssl/sslcontext/sslcontext :

import socket
import ssl

HOST = "127.0.0.1"
PORT = 8443

if __name__ == "__main__":
    context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    context.load_cert_chain(certfile="/path/to/certfile", keyfile="/path/to/keyfile")
    context.load_verify_locations(cafile="/path/to/certfile")

    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server = context.wrap_socket(s)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

    s.close()

    server.bind((HOST, PORT))
    server.listen(0)

    while True:
        connection, client_address = server.accept()
        while True:
            data = connection.recv(1024)
            if not data:
                break
            print(f"Received: {data.decode('utf-8')}")
import socket
import ssl
import time

HOST = "127.0.0.1"
PORT = 8443

if __name__ == "__main__":
    context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    context.load_cert_chain(certfile="/path/to/certfile", keyfile="/path/to/keyfile")
    context.load_verify_locations(cafile="/path/to/certfile")
    
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client = context.wrap_socket(s, server_hostname=HOST)
    s.close()

    client.connect((HOST, PORT))

    while True:
        client.sendall("Hello World!".encode("utf-8"))
        time.sleep(1)

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment