.
, ssh .
   , .

:
, , , 9999.
512 ţ escda .
, /tmp/_Good_check_ .
.

:
, , .
, , , .

/tmp/_Good_check_ ,
- sshd, - sshd, .
firejail- - , , , .

#==========================================>server:

#!/usr/bin/env python3
import sqlite3
import socket
import os
import tempfile
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes

def init_db():
    conn = sqlite3.connect('server.db')
    c = conn.cursor()
    c.execute('''CREATE TABLE IF NOT EXISTS server_config (id INTEGER PRIMARY KEY, public_key BLOB)''')
    conn.commit()
    conn.close()

def get_public_key():
    conn = sqlite3.connect('server.db')
    c = conn.cursor()
    c.execute("SELECT public_key FROM server_config LIMIT 1")
    result = c.fetchone()
    conn.close()
    return result[0] if result else None

def main():
    init_db()
    public_key_bytes = get_public_key()
    
    if public_key_bytes is None:
        print("No public key found. Enter client's public key (DER hex):")
        user_input = input().strip()
        try:
            public_key_bytes = bytes.fromhex(user_input)
            if len(public_key_bytes) < 10:
                raise ValueError("Invalid public key length")
            conn = sqlite3.connect('server.db')
            c = conn.cursor()
            c.execute("INSERT INTO server_config (public_key) VALUES (?)", (public_key_bytes,))
            conn.commit()
            conn.close()
            print("Public key stored.")
        except ValueError as e:
            print(f"Invalid hex input: {e}. Exiting.")
            exit(1)
#ipv4
#    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
#    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
#    server_socket.bind(('0.0.0.0', 9999)) #ipv4
#    server_socket.listen()
#ipv6
#    server_socket = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
#    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
#    server_socket.bind(('::', 9999))  # :: - IPv6
#    server_socket.listen()
#universal AF_INET6 IPV6_V6ONLY = 0
    server_socket = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_socket.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0)  # IPv4 IPv6
    server_socket.bind(('::', 9999))
    server_socket.listen()

    print("Server listening on port 9999...")
    print("Server will reject connections if signature exceeds 80 bytes")
    
    while True:
        try:
            client_socket, addr = server_socket.accept()
            print(f"Connection from {addr}")
            
            challenge = os.urandom(512)
            client_socket.sendall(challenge)
            
            # 80 # 70 72 , , 80.
            signature = b''
            total_received = 0
            max_signature_size = 80
            
            try:
                while total_received < max_signature_size:
                    chunk = client_socket.recv(1024)
                    if not chunk:
                        break
                    signature += chunk
                    total_received += len(chunk)
                    
                    #
                    if total_received >= max_signature_size:
                        print(f"Signature size limit exceeded: {total_received} bytes")
                        client_socket.close()
                        break
                        
            except Exception as e:
                print(f"Error receiving signature: {e}")
                client_socket.close()
                continue
            
            #
            client_socket.close()
            
            #
            if total_received > max_signature_size:
                print(f"Signature too large: {total_received} bytes (limit: {max_signature_size})")
                continue
            
            # ,
            if len(signature) == 0:
                print("Empty signature received")
                continue
            
            print(f"Received signature: {len(signature)} bytes")
            
            try:
                public_key = serialization.load_der_public_key(public_key_bytes)
                public_key.verify(signature, challenge, ec.ECDSA(hashes.SHA256()))

                if os.path.isfile('/tmp/_Good_check_'):
                # -
                    os.remove('/tmp/_Good_check_')
                    print(f" /tmp/_Good_check_ .")
                else:
                # -
                    with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.check') as f:
                      f.write('Signature valid')
                      temp_path = f.name
                    os.rename(temp_path, '/tmp/_Good_check_')
                    print(f" /tmp/_Good_check_ .")

                print("Signature verified successfully")
            except Exception as e:
                print(f"Verification failed: {str(e)}")
                
        except Exception as e:
            print(f"Error handling connection: {e}")
            continue

if __name__ == '__main__':
    main()

#============================================>client:

#!/usr/bin/env python3

import sqlite3
import socket
import os
from cryptography.hazmat.primitives import serialization, hashes
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.backends import default_backend

def init_db():
    conn = sqlite3.connect('client.db')
    c = conn.cursor()
    c.execute('''CREATE TABLE IF NOT EXISTS client_keys (id INTEGER PRIMARY KEY, private_key BLOB, public_key BLOB)''')
    conn.commit()
    conn.close()

def get_client_keys():
    conn = sqlite3.connect('client.db')
    c = conn.cursor()
    c.execute("SELECT private_key, public_key FROM client_keys LIMIT 1")
    result = c.fetchone()
    conn.close()
    return result if result else (None, None)

def generate_keys():
    priv_key = ec.generate_private_key(ec.SECP256K1(), backend=default_backend())
    pub_key = priv_key.public_key()
    
    private_bytes = priv_key.private_bytes(
        encoding=serialization.Encoding.DER,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=serialization.NoEncryption()
    )
    public_bytes = pub_key.public_bytes(
        encoding=serialization.Encoding.DER,
        format=serialization.PublicFormat.SubjectPublicKeyInfo
    )
    
    conn = sqlite3.connect('client.db')
    c = conn.cursor()
    c.execute("INSERT INTO client_keys (private_key, public_key) VALUES (?, ?)", (private_bytes, public_bytes))
    conn.commit()
    conn.close()
    
    print(f"Public key (DER hex): {public_bytes.hex()}")

def main():
    init_db()
    private_key_bytes, public_key_bytes = get_client_keys()
    
    if private_key_bytes is None:
        generate_keys()
        private_key_bytes, public_key_bytes = get_client_keys()
    
    try:
        private_key = serialization.load_der_private_key(
            private_key_bytes,
            password=None,
            backend=default_backend()
        )
    except Exception as e:
        print(f"Key load failed: {str(e)}")
        exit(1)
    
#ipv4
#    client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
#    client_socket.connect(('localhost', 9999))
#ipv6
#    client_socket = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
#    client_socket.connect(('x:x:x:x:x:x:x:x', 9999))
#universal
    addr_info = socket.getaddrinfo('127.0.0.1', 9999, socket.AF_UNSPEC, socket.SOCK_STREAM)
    client_socket = socket.socket(addr_info[0][0], addr_info[0][1])
    client_socket.connect(addr_info[0][4])

    
    challenge = client_socket.recv(4096)
    if not challenge:
        print("No challenge received.")
        exit(1)
    
    try:
        signature = private_key.sign(challenge, ec.ECDSA(hashes.SHA256()))
    except Exception as e:
        print(f"Signing failed: {str(e)}")
        exit(1)
    
    client_socket.sendall(signature)    #sig
#    client_socket.sendall(challenge)    #trash
    client_socket.close()
    sig_len=len(signature)
    print("Signature sent. Sig_len:", sig_len)

#    chall = challenge.hex()
#    print("c:",chall)
#    sig = signature.hex()
#    print("s:",sig)

if __name__ == '__main__':
    main()