1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
|
"""
Self-signed TLS certificate generation for the node.
The cert is used for transport confidentiality only.
Node identity is verified via Ed25519 PK (from hub), not TLS cert chain.
Clients connect with ssl.CERT_NONE + verify Ed25519 at the MNP handshake layer.
Certificate is generated once and cached at ~/.config/meshbay/node_tls.pem/.key.
"""
import logging
import os
from pathlib import Path
import ssl
import datetime
import ipaddress
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.x509.oid import NameOID
log = logging.getLogger(__name__)
DEFAULT_CERT = Path.home() / ".config" / "meshbay" / "node_tls.crt"
DEFAULT_KEY = Path.home() / ".config" / "meshbay" / "node_tls.key"
def generate_self_signed_cert(
cert_path: Path = DEFAULT_CERT,
key_path: Path = DEFAULT_KEY,
) -> tuple[Path, Path]:
"""Generate a self-signed RSA-2048 TLS cert valid for 10 years."""
cert_path.parent.mkdir(parents=True, exist_ok=True)
rsa_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
subject = issuer = x509.Name([
x509.NameAttribute(NameOID.COMMON_NAME, "meshbay-node"),
])
cert = (
x509.CertificateBuilder()
.subject_name(subject)
.issuer_name(issuer)
.public_key(rsa_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=3650))
.add_extension(
x509.SubjectAlternativeName([
x509.DNSName("localhost"),
x509.IPAddress(ipaddress.IPv4Address("127.0.0.1")),
]),
critical=False,
)
.sign(rsa_key, hashes.SHA256())
)
cert_path.write_bytes(cert.public_bytes(serialization.Encoding.PEM))
key_path.write_bytes(rsa_key.private_bytes(
serialization.Encoding.PEM,
serialization.PrivateFormat.TraditionalOpenSSL,
serialization.NoEncryption(),
))
cert_path.chmod(0o644)
key_path.chmod(0o600)
log.info("TLS cert generated: %s", cert_path)
return cert_path, key_path
def server_ssl_context(
cert_path: Path = DEFAULT_CERT,
key_path: Path = DEFAULT_KEY,
) -> ssl.SSLContext:
"""SSL context for the node's TCP server."""
if not cert_path.exists() or not key_path.exists():
generate_self_signed_cert(cert_path, key_path)
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
ctx.load_cert_chain(certfile=cert_path, keyfile=key_path)
ctx.minimum_version = ssl.TLSVersion.TLSv1_3
return ctx
def client_ssl_context() -> ssl.SSLContext:
"""
SSL context for clients connecting to a node.
CERT_NONE because we verify node identity via Ed25519 PK at the MNP layer.
"""
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
ctx.minimum_version = ssl.TLSVersion.TLSv1_3
return ctx
|