/
/
/
1"""
2WebRTC DTLS Certificate Management.
3
4This module provides persistent DTLS certificate management for WebRTC connections.
5The certificate is generated once and stored persistently, enabling client-side
6certificate pinning for authentication.
7"""
8
9from __future__ import annotations
10
11import base64
12import logging
13import os
14import stat
15from datetime import timedelta
16from pathlib import Path
17
18from cryptography import x509
19from cryptography.hazmat.primitives import hashes, serialization
20from cryptography.hazmat.primitives.asymmetric import ec
21from cryptography.x509.oid import NameOID
22
23from music_assistant.helpers.datetime import utc
24
25LOGGER = logging.getLogger(__name__)
26
27CERT_FILENAME = "webrtc_certificate.pem"
28KEY_FILENAME = "webrtc_private_key.pem"
29
30CERT_VALIDITY_DAYS = 3650 # 10 years
31
32CERT_RENEWAL_THRESHOLD_DAYS = 30
33
34
35def _generate_certificate() -> tuple[ec.EllipticCurvePrivateKey, x509.Certificate]:
36 """
37 Generate a new ECDSA certificate for WebRTC DTLS.
38
39 :return: Tuple of (private_key, certificate).
40 """
41 # Generate ECDSA key (SECP256R1 - the standard WebRTC DTLS curve)
42 private_key = ec.generate_private_key(ec.SECP256R1())
43
44 now = utc()
45 not_before = now - timedelta(days=1)
46 not_after = now + timedelta(days=CERT_VALIDITY_DAYS)
47
48 subject = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "Music Assistant WebRTC")])
49
50 cert = (
51 x509.CertificateBuilder()
52 .subject_name(subject)
53 .issuer_name(subject)
54 .public_key(private_key.public_key())
55 .serial_number(x509.random_serial_number())
56 .not_valid_before(not_before)
57 .not_valid_after(not_after)
58 .sign(private_key, hashes.SHA256())
59 )
60
61 return private_key, cert
62
63
64def _save_certificate(
65 storage_path: str,
66 private_key: ec.EllipticCurvePrivateKey,
67 cert: x509.Certificate,
68) -> None:
69 """
70 Save certificate and private key to disk.
71
72 :param storage_path: Directory to store the files.
73 :param private_key: The EC private key.
74 :param cert: The X.509 certificate.
75 """
76 cert_path = Path(storage_path) / CERT_FILENAME
77 key_path = Path(storage_path) / KEY_FILENAME
78
79 cert_pem = cert.public_bytes(serialization.Encoding.PEM)
80 cert_path.write_bytes(cert_pem)
81
82 key_pem = private_key.private_bytes(
83 encoding=serialization.Encoding.PEM,
84 format=serialization.PrivateFormat.PKCS8,
85 encryption_algorithm=serialization.NoEncryption(),
86 )
87 fd = os.open(key_path, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o600)
88 with os.fdopen(fd, "wb") as f:
89 # O_TRUNC keeps a pre-existing file's mode, so tighten permissions on the
90 # fd before any key byte is written (owner read/write only)
91 os.fchmod(fd, stat.S_IRUSR | stat.S_IWUSR)
92 f.write(key_pem)
93
94
95def _load_certificate(
96 storage_path: str,
97) -> tuple[ec.EllipticCurvePrivateKey, x509.Certificate] | None:
98 """
99 Load certificate and private key from disk.
100
101 :param storage_path: Directory containing the files.
102 :return: Tuple of (private_key, certificate) or None if files don't exist.
103 """
104 cert_path = Path(storage_path) / CERT_FILENAME
105 key_path = Path(storage_path) / KEY_FILENAME
106
107 if not cert_path.exists() or not key_path.exists():
108 return None
109
110 try:
111 cert_pem = cert_path.read_bytes()
112 cert = x509.load_pem_x509_certificate(cert_pem)
113
114 key_pem = key_path.read_bytes()
115 private_key = serialization.load_pem_private_key(key_pem, password=None)
116
117 if not isinstance(private_key, ec.EllipticCurvePrivateKey):
118 LOGGER.warning("WebRTC private key is not an EC key, will regenerate")
119 return None
120
121 # cert and key are written as two separate files; a crash between the writes
122 # leaves a mismatched pair that would otherwise fail every DTLS handshake
123 if private_key.public_key() != cert.public_key():
124 LOGGER.warning("WebRTC private key does not match certificate, will regenerate")
125 return None
126
127 # older versions wrote the key with umask permissions before chmoding it;
128 # a crash in that window left it world-readable, and a valid pair is never
129 # rewritten, so repair permissions on load (owner read/write only)
130 mode = stat.S_IRUSR | stat.S_IWUSR
131 if stat.S_IMODE(key_path.stat().st_mode) != mode:
132 key_path.chmod(mode)
133
134 return private_key, cert
135 except Exception as err:
136 LOGGER.warning("Failed to load WebRTC certificate: %s", err)
137 return None
138
139
140def _is_certificate_valid(cert: x509.Certificate) -> bool:
141 """
142 Check if certificate is still valid with enough time remaining.
143
144 :param cert: The X.509 certificate to check.
145 :return: True if certificate is valid and has sufficient time remaining.
146 """
147 now = utc()
148 not_after = cert.not_valid_after_utc
149
150 if now >= not_after:
151 return False
152
153 days_remaining = (not_after - now).days
154 return not days_remaining < CERT_RENEWAL_THRESHOLD_DAYS
155
156
157def _get_or_create_certificate(
158 storage_path: str,
159) -> tuple[ec.EllipticCurvePrivateKey, x509.Certificate]:
160 """
161 Load a valid persisted DTLS keypair, or generate and persist a new one.
162
163 :param storage_path: Directory to store/load the certificate files.
164 :return: Tuple of (private_key, certificate).
165 """
166 loaded = _load_certificate(storage_path)
167 if loaded is not None and _is_certificate_valid(loaded[1]):
168 return loaded
169
170 LOGGER.debug("Generating new WebRTC DTLS certificate (valid for %d days)", CERT_VALIDITY_DAYS)
171 private_key, cert = _generate_certificate()
172 _save_certificate(storage_path, private_key, cert)
173 return private_key, cert
174
175
176def _remote_id_from_certificate(cert: x509.Certificate) -> str:
177 """
178 Derive the deterministic Remote ID from a certificate.
179
180 :param cert: The X.509 certificate to derive the Remote ID from.
181 :return: Custom base32-encoded (with 9s instead of 2s) Remote ID string
182 (26 characters, uppercase, no-padding).
183 """
184 # SHA-256 over the DER certificate, matching the DTLS certificate digest so Remote IDs
185 # stay stable; take the first 128 bits and base32-encode (with 9s instead of 2s), no padding.
186 digest = cert.fingerprint(hashes.SHA256())
187 return base64.b32encode(digest[:16]).decode("ascii").rstrip("=").replace("2", "9")
188
189
190def get_or_create_remote_id(storage_path: str) -> str:
191 """
192 Return the stable Remote ID for this instance without loading the WebRTC lib.
193
194 Loads (or creates and persists) the WebRTC DTLS certificate and derives the
195 Remote ID from it, so the always-on remote_access/info endpoint can report the
196 Remote ID even when remote access is disabled and the native lib was never loaded.
197
198 :param storage_path: Directory to store/load the certificate files.
199 :return: The Remote ID derived from the persistent certificate.
200 """
201 _, cert = _get_or_create_certificate(storage_path)
202 return _remote_id_from_certificate(cert)
203
204
205def get_or_create_webrtc_certificate_pems(storage_path: str) -> tuple[str, str]:
206 """
207 Get or create the persistent WebRTC DTLS certificate as PEM strings.
208
209 :param storage_path: Directory to store/load the certificate files.
210 :return: Tuple of (certificate_pem, private_key_pem).
211 """
212 private_key, cert = _get_or_create_certificate(storage_path)
213 cert_pem = cert.public_bytes(serialization.Encoding.PEM).decode()
214 key_pem = private_key.private_bytes(
215 encoding=serialization.Encoding.PEM,
216 format=serialization.PrivateFormat.PKCS8,
217 encryption_algorithm=serialization.NoEncryption(),
218 ).decode()
219 return cert_pem, key_pem
220