a2258d3df4
- packages/p2p: identity (peer_id from sha256 of RSA pubkey), TLS cert generation with SHA-256 fingerprint, GossipClient (WSS PubSub with per-topic handlers, dedup by msg_id, auto-reconnect), MdnsDiscovery (zeroconf optional dependency, graceful no-op fallback) - packages/relay: new meshnet-relay package — RelayServer (asyncio + websockets) with gossip fanout hub, circuit relay proxy for NAT traversal, peer registry; meshnet-relay CLI - packages/p2p/relay_bootstrap.json: team relay bootstrap list - Tracker: _NodeEntry gains relay_addr, cert_fingerprint, peer_id; both register and heartbeat handlers read and store these optional fields - docs/adr/0010 already written (previous commit) - conftest.py: packages/relay added to sys.path - 18 new tests; 115 passed total, 1 skipped (no regressions) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
115 lines
3.6 KiB
Python
115 lines
3.6 KiB
Python
"""TLS certificate generation and fingerprint helpers for node-to-node comms."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime
|
|
import hashlib
|
|
import ipaddress
|
|
import json
|
|
import os
|
|
import socket
|
|
import ssl
|
|
import stat
|
|
from pathlib import Path
|
|
|
|
_CERT_PATH = Path.home() / ".config" / "meshnet" / "node_cert.pem"
|
|
_KEY_PATH = Path.home() / ".config" / "meshnet" / "node_key.pem"
|
|
|
|
|
|
def generate_self_signed_cert(
|
|
cert_path: Path | None = None,
|
|
key_path: Path | None = None,
|
|
common_name: str | None = None,
|
|
) -> tuple[Path, Path]:
|
|
"""Generate a self-signed RSA-2048 cert valid for 10 years.
|
|
|
|
Returns (cert_path, key_path). Skips generation if both files already exist.
|
|
"""
|
|
from cryptography import x509
|
|
from cryptography.hazmat.primitives import hashes, serialization
|
|
from cryptography.hazmat.primitives.asymmetric import rsa
|
|
from cryptography.x509.oid import NameOID
|
|
|
|
cert_p = cert_path or _CERT_PATH
|
|
key_p = key_path or _KEY_PATH
|
|
|
|
if cert_p.exists() and key_p.exists():
|
|
return cert_p, key_p
|
|
|
|
cert_p.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
cn = common_name or socket.getfqdn()
|
|
|
|
subject = issuer = x509.Name([
|
|
x509.NameAttribute(NameOID.COMMON_NAME, cn),
|
|
x509.NameAttribute(NameOID.ORGANIZATION_NAME, "meshnet-node"),
|
|
])
|
|
|
|
san_list: list = [x509.DNSName(cn)]
|
|
try:
|
|
san_list.append(x509.IPAddress(ipaddress.IPv4Address(socket.gethostbyname(cn))))
|
|
except (socket.gaierror, ValueError):
|
|
pass
|
|
san_list.append(x509.IPAddress(ipaddress.IPv4Address("127.0.0.1")))
|
|
|
|
cert = (
|
|
x509.CertificateBuilder()
|
|
.subject_name(subject)
|
|
.issuer_name(issuer)
|
|
.public_key(key.public_key())
|
|
.serial_number(x509.random_serial_number())
|
|
.not_valid_before(datetime.datetime.now(datetime.timezone.utc))
|
|
.not_valid_after(datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(days=365 * 10))
|
|
.add_extension(x509.SubjectAlternativeName(san_list), critical=False)
|
|
.sign(key, hashes.SHA256())
|
|
)
|
|
|
|
key_pem = key.private_bytes(
|
|
serialization.Encoding.PEM,
|
|
serialization.PrivateFormat.TraditionalOpenSSL,
|
|
serialization.NoEncryption(),
|
|
)
|
|
cert_pem = cert.public_bytes(serialization.Encoding.PEM)
|
|
|
|
key_p.write_bytes(key_pem)
|
|
cert_p.write_bytes(cert_pem)
|
|
try:
|
|
os.chmod(key_p, stat.S_IRUSR | stat.S_IWUSR)
|
|
except OSError:
|
|
pass
|
|
|
|
return cert_p, key_p
|
|
|
|
|
|
def cert_fingerprint(cert_path: Path | None = None) -> str:
|
|
"""Return sha256 fingerprint of the cert as 'sha256:<hex>'."""
|
|
from cryptography import x509
|
|
from cryptography.hazmat.primitives import hashes
|
|
|
|
p = cert_path or _CERT_PATH
|
|
cert = x509.load_pem_x509_certificate(p.read_bytes())
|
|
fp = cert.fingerprint(hashes.SHA256()).hex()
|
|
return f"sha256:{fp}"
|
|
|
|
|
|
def make_server_ssl_context(
|
|
cert_path: Path | None = None,
|
|
key_path: Path | None = None,
|
|
) -> ssl.SSLContext:
|
|
"""Return an ssl.SSLContext for a server using our self-signed cert."""
|
|
cert_p = cert_path or _CERT_PATH
|
|
key_p = key_path or _KEY_PATH
|
|
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
|
|
ctx.load_cert_chain(certfile=str(cert_p), keyfile=str(key_p))
|
|
return ctx
|
|
|
|
|
|
def make_client_ssl_context(verify: bool = False) -> ssl.SSLContext:
|
|
"""Return a client SSLContext. verify=False for self-signed TOFU connections."""
|
|
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
|
|
if not verify:
|
|
ctx.check_hostname = False
|
|
ctx.verify_mode = ssl.CERT_NONE
|
|
return ctx
|