Simple TLS client and server on python

tls_client.py

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)

tls_server.py

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')}")

You can see my project https://github.com/NguyenKhue09/Socket-Server

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.

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')}")

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)