lib.tls

  1import base64
  2import hashlib
  3import json
  4import re
  5import shlex
  6import ssl
  7from dataclasses import dataclass, field
  8from ipaddress import IPv4Interface, IPv6Interface
  9from datetime import datetime
 10from urllib.parse import urlparse
 11
 12from SRE.lib_sre import Grade0, NetScheme0
 13
 14
 15def _cert_fingerprint_sha256(pem: str) -> str:
 16    """Return the SHA256 fingerprint of a PEM certificate as XX:XX:... uppercase hex."""
 17    der = ssl.PEM_cert_to_DER_cert(pem.strip())
 18    digest = hashlib.sha256(der).hexdigest()
 19    return ':'.join(digest[i:i + 2].upper() for i in range(0, len(digest), 2))
 20
 21
 22def _val(line, prefix):
 23    # OpenSSL 3 prints "sha256 Fingerprint=", OpenSSL 1.1 "SHA256 Fingerprint=": ignore the case.
 24    m = re.search(rf'{prefix}=(.+)', line, flags=re.IGNORECASE)
 25    return m.group(1).strip() if m else ''
 26
 27
 28def _cert_dict(subject_out, issuer_out, dates_out, serial_out, fingerprint_out):
 29    not_before_m = re.search(r'notBefore=(.+)', dates_out)
 30    not_after_m  = re.search(r'notAfter=(.+)',  dates_out)
 31    subject = _val(subject_out, "subject")
 32    cn_m = re.search(r'(?:^|[,/]\s*)CN\s*=\s*([^,/]+)', subject)
 33    return {
 34        "subject":      subject,
 35        "issuer":       _val(issuer_out,      "issuer"),
 36        "common_name":  cn_m.group(1).strip() if cn_m else '',
 37        "not_before":   not_before_m.group(1).strip() if not_before_m else '',
 38        "not_after":    not_after_m.group(1).strip()  if not_after_m  else '',
 39        "serial":       _val(serial_out,      "serial"),
 40        "fingerprint":  _val(fingerprint_out, "SHA256 Fingerprint"),
 41    }
 42
 43
 44def eval_rsa_private_key(grade: Grade0, machine_name: str, key_file: str,
 45                         password: str | None = None, bits: int = 4096,
 46                         cipher: str = "AES-256-CBC",
 47                         step: int = 1) -> bool:
 48    """Check that key_file is an RSA private key with the expected properties.
 49
 50    Verifies:
 51    - The file is an RSA private key (decryptable with password if provided).
 52    - The key size matches bits.
 53    - The encryption cipher in the PEM header matches cipher (case-insensitive).
 54      This applies to traditional PEM format (BEGIN RSA PRIVATE KEY); for
 55      PKCS#8 (BEGIN ENCRYPTED PRIVATE KEY) the cipher check is skipped.
 56      The cipher check is also skipped when password is None.
 57
 58    Args:
 59        grade:        the Grade0 instance.
 60        machine_name: name of the virtual machine to inspect.
 61        key_file:     absolute path to the private key file on the machine.
 62        password:     passphrase protecting the private key, or None if unencrypted.
 63        bits:         expected RSA key size (default: 4096).
 64        cipher:       expected PEM encryption cipher (default: 'AES-256-CBC').
 65        step:         step number passed to grade.test() (default: 1).
 66
 67    Returns:
 68        True if all checks pass, False otherwise.
 69    """
 70    passin = f"-passin {shlex.quote(f'pass:{password}')}" if password is not None else ""
 71    q_key_file = shlex.quote(key_file)
 72    key_text, key_code = grade.test(
 73        machine_name=machine_name,
 74        command=f"openssl rsa -in {q_key_file} {passin} -noout -text 2>&1",
 75        step=step,
 76        allow_error=True,
 77    )
 78    dek_info, _ = grade.test(
 79        machine_name=machine_name,
 80        command=f"grep 'DEK-Info' {q_key_file}",
 81        step=step,
 82        allow_error=True,
 83    )
 84
 85    if key_code != 0:
 86        return False
 87
 88    bits_m = re.search(r'Private-Key:\s*\((\d+)\s*bit', key_text)
 89    if not bits_m or int(bits_m.group(1)) != bits:
 90        return False
 91
 92    # DEK-Info line only present in traditional PEM format; skip cipher check for PKCS#8
 93    if dek_info.strip():
 94        dek_m = re.search(r'DEK-Info:\s*([^,\s]+)', dek_info)
 95        if not dek_m or dek_m.group(1).upper() != cipher.upper():
 96            return False
 97
 98    return True
 99
100
101def set_rsa_private_key(net_scheme: NetScheme0, machine_name: str, key_file: str,
102                        password: str, bits: int = 4096,
103                        cipher: str = "AES-256-CBC"):
104    """Generate an RSA private key on machine_name.
105
106    Args:
107        net_scheme:   the NetScheme0 instance.
108        machine_name: name of the virtual machine.
109        key_file:     absolute path where the key will be written on the machine.
110        password:     passphrase to protect the key.
111        bits:         RSA key size in bits (default: 4096).
112        cipher:       PEM encryption cipher (default: 'AES-256-CBC').
113    """
114    net_scheme.cmd(machine_name,
115                   f"openssl genrsa -{shlex.quote(cipher.lower())} -passout {shlex.quote(f'pass:{password}')} -out {shlex.quote(key_file)} {bits}")
116
117
118def eval_self_signed_certificate(grade: Grade0, machine_name: str,
119                                 key_file: str, cert_file: str,
120                                 password: str,
121                                 step: int = 1) -> dict | None:
122    """Check that key_file and cert_file are a password-protected TLS key and
123    a matching self-signed certificate on machine_name.
124
125    Verifies:
126    - key_file is a valid PEM private key decryptable with password.
127    - cert_file is a valid PEM certificate whose Issuer equals its Subject
128      (self-signed).
129    - The public key in the certificate matches the private key.
130
131    Args:
132        grade:        the Grade0 instance.
133        machine_name: name of the virtual machine to inspect.
134        key_file:     absolute path to the private key file on the machine.
135        cert_file:    absolute path to the certificate file on the machine.
136        password:     passphrase protecting the private key.
137        step:         step number passed to grade.test() (default: 1).
138
139    Returns:
140        A dict with certificate fields (subject, issuer, not_before, not_after,
141        serial, fingerprint) if all checks pass, None otherwise.
142    """
143    # All grade.test() calls must be made unconditionally so they are registered
144    # in the first (registration) pass and carry real results in the second pass.
145    q_key_file = shlex.quote(key_file)
146    q_cert_file = shlex.quote(cert_file)
147    q_passin = shlex.quote(f'pass:{password}')
148    _, key_code = grade.test(
149        machine_name=machine_name,
150        command=f"openssl pkey -in {q_key_file} -passin {q_passin} -noout",
151        step=step,
152        allow_error=True,
153    )
154    cert_text, cert_code = grade.test(
155        machine_name=machine_name,
156        command=f"openssl x509 -in {q_cert_file} -noout -text -fingerprint -sha256",
157        step=step,
158        allow_error=True,
159    )
160    cert_pubkey, cert_pubkey_code = grade.test(
161        machine_name=machine_name,
162        command=f"openssl x509 -in {q_cert_file} -noout -pubkey",
163        step=step,
164        allow_error=True,
165    )
166    key_pubkey, key_pubkey_code = grade.test(
167        machine_name=machine_name,
168        command=f"openssl pkey -in {q_key_file} -passin {q_passin} -pubout",
169        step=step,
170        allow_error=True,
171    )
172    subject_out, _ = grade.test(
173        machine_name=machine_name,
174        command=f"openssl x509 -in {q_cert_file} -noout -subject",
175        step=step,
176        allow_error=True,
177    )
178    issuer_out, _ = grade.test(
179        machine_name=machine_name,
180        command=f"openssl x509 -in {q_cert_file} -noout -issuer",
181        step=step,
182        allow_error=True,
183    )
184    dates_out, _ = grade.test(
185        machine_name=machine_name,
186        command=f"openssl x509 -in {q_cert_file} -noout -dates",
187        step=step,
188        allow_error=True,
189    )
190    serial_out, _ = grade.test(
191        machine_name=machine_name,
192        command=f"openssl x509 -in {q_cert_file} -noout -serial",
193        step=step,
194        allow_error=True,
195    )
196    fingerprint_out, _ = grade.test(
197        machine_name=machine_name,
198        command=f"openssl x509 -in {q_cert_file} -noout -fingerprint -sha256",
199        step=step,
200        allow_error=True,
201    )
202
203    if key_code != 0 or cert_code != 0 or cert_pubkey_code != 0 or key_pubkey_code != 0:
204        return None
205
206    if cert_pubkey.strip() != key_pubkey.strip():
207        return None
208
209    # Check self-signed: Subject must equal Issuer
210    subject_m = re.search(r'Subject:\s*(.+)', cert_text)
211    issuer_m  = re.search(r'Issuer:\s*(.+)',  cert_text)
212    if not subject_m or not issuer_m:
213        return None
214    if subject_m.group(1).strip() != issuer_m.group(1).strip():
215        return None
216
217    return _cert_dict(subject_out, issuer_out, dates_out, serial_out, fingerprint_out)
218
219
220def eval_certificate(grade: Grade0, machine_name: str,
221                     key_file: str, cert_file: str,
222                     step: int = 1) -> dict | None:
223    """Check that key_file and cert_file are a matching TLS key pair on machine_name.
224
225    Verifies:
226    - cert_file is a valid PEM certificate.
227    - The public key in the certificate matches the private key in key_file.
228
229    Args:
230        grade:        the Grade0 instance.
231        machine_name: name of the virtual machine to inspect.
232        key_file:     absolute path to the private key file on the machine.
233        cert_file:    absolute path to the certificate file on the machine.
234        step:         step number passed to grade.test() (default: 1).
235
236    Returns:
237        A dict with certificate fields (subject, issuer, common_name,
238        not_before, not_after, serial, fingerprint) if all checks pass,
239        None otherwise.
240    """
241    # All grade.test() calls must be made unconditionally so they are registered
242    # in the first (registration) pass and carry real results in the second pass.
243    q_key_file = shlex.quote(key_file)
244    q_cert_file = shlex.quote(cert_file)
245    _, cert_code = grade.test(
246        machine_name=machine_name,
247        command=f"openssl x509 -in {q_cert_file} -noout -text -fingerprint -sha256",
248        step=step,
249        allow_error=True,
250    )
251    cert_pubkey, cert_pubkey_code = grade.test(
252        machine_name=machine_name,
253        command=f"openssl x509 -in {q_cert_file} -noout -pubkey",
254        step=step,
255        allow_error=True,
256    )
257    key_pubkey, key_pubkey_code = grade.test(
258        machine_name=machine_name,
259        command=f"openssl pkey -in {q_key_file} -pubout",
260        step=step,
261        allow_error=True,
262    )
263    subject_out, _ = grade.test(
264        machine_name=machine_name,
265        command=f"openssl x509 -in {q_cert_file} -noout -subject",
266        step=step,
267        allow_error=True,
268    )
269    issuer_out, _ = grade.test(
270        machine_name=machine_name,
271        command=f"openssl x509 -in {q_cert_file} -noout -issuer",
272        step=step,
273        allow_error=True,
274    )
275    dates_out, _ = grade.test(
276        machine_name=machine_name,
277        command=f"openssl x509 -in {q_cert_file} -noout -dates",
278        step=step,
279        allow_error=True,
280    )
281    serial_out, _ = grade.test(
282        machine_name=machine_name,
283        command=f"openssl x509 -in {q_cert_file} -noout -serial",
284        step=step,
285        allow_error=True,
286    )
287    fingerprint_out, _ = grade.test(
288        machine_name=machine_name,
289        command=f"openssl x509 -in {q_cert_file} -noout -fingerprint -sha256",
290        step=step,
291        allow_error=True,
292    )
293
294    if cert_code != 0 or cert_pubkey_code != 0 or key_pubkey_code != 0:
295        return None
296
297    if cert_pubkey.strip() != key_pubkey.strip():
298        return None
299
300    return _cert_dict(subject_out, issuer_out, dates_out, serial_out, fingerprint_out)
301
302
303def eval_certificate_validity(grade: Grade0, machine_name: str,
304                              cert_file: str, ca_cert_file: str,
305                              step: int = 1, untrusted: str | None = None) -> bool:
306    """Check that cert_file is signed by ca_cert_file on machine_name.
307
308    Args:
309        grade:        the Grade0 instance.
310        machine_name: name of the virtual machine to inspect.
311        cert_file:    absolute path to the certificate file on the machine.
312        ca_cert_file: absolute path to the CA certificate file on the machine.
313        step:         step number passed to grade.test() (default: 1).
314        untrusted:    optional PEM file of intermediate certificates (``-untrusted``),
315                      e.g. the leaf bundle written by step-ca whose second block is the
316                      intermediate CA.
317
318    Returns:
319        True if cert_file is a valid certificate chaining to ca_cert_file,
320        False otherwise.
321    """
322    untrusted_opt = f" -untrusted {shlex.quote(untrusted)}" if untrusted else ""
323    _, verify_code = grade.test(
324        machine_name=machine_name,
325        command=f"openssl verify -CAfile {shlex.quote(ca_cert_file)}{untrusted_opt} {shlex.quote(cert_file)}",
326        step=step,
327        allow_error=True,
328    )
329    return verify_code == 0
330
331
332def _bracketed(ip) -> str:
333    """The address for curl / openssl: an IPv6 address goes between brackets.
334
335    *ip* may be a string or an ``ipaddress`` Address / Interface object (prefix stripped).
336    """
337    host = str(ip.ip) if isinstance(ip, (IPv4Interface, IPv6Interface)) else str(ip)
338    return f'[{host}]' if ':' in host else host
339
340
341def _host_port(ip, port: int) -> str:
342    """``host:port`` for curl / openssl, with the brackets an IPv6 address needs."""
343    return f'{_bracketed(ip)}:{port}'
344
345
346def eval_https_server(grade: Grade0, machine_name: str, url: str,
347                      server_ip, cert: str,
348                      server_port: int = 443, step: int = 1) -> bool:
349    """Check that an HTTPS server at server_ip responds correctly and presents
350    the expected certificate.
351
352    Verifies:
353    - A GET request to url (routed to server_ip:server_port) succeeds (2xx).
354    - The certificate presented by the server matches cert.
355
356    Args:
357        grade:        the Grade0 instance.
358        machine_name: name of the virtual machine from which to connect.
359        url:          full URL to request (e.g. https://myserver/index.html).
360        server_ip:    IP address of the HTTPS server to connect to (IPv4 or IPv6, as a
361                      string or an ``ipaddress`` object; IPv6 is bracketed automatically).
362        cert:         PEM certificate content to verify against the server.
363        server_port:  HTTPS port (default: 443).
364        step:         step number passed to grade.test() (default: 1).
365
366    Returns:
367        True if all checks pass, False otherwise.
368    """
369    # All grade.test() calls must be made unconditionally so they are registered
370    # in the first (registration) pass and carry real results in the second pass.
371    _, http_code = grade.test(
372        machine_name=machine_name,
373        command=f"curl -k -L --fail --connect-to ::{_host_port(server_ip, server_port)}"
374                f" -s -o /dev/null {url}",
375        step=step,
376        allow_error=True,
377    )
378    server_fp, server_fp_code = grade.test(
379        machine_name=machine_name,
380        command=f"openssl s_client -connect {_host_port(server_ip, server_port)}"
381                f" </dev/null 2>/dev/null | openssl x509 -noout -fingerprint -sha256",
382        step=step,
383        allow_error=True,
384    )
385
386    if http_code != 0:
387        return False
388    if server_fp_code != 0:
389        return False
390
391    try:
392        cert_fp_val = _cert_fingerprint_sha256(cert)
393    except Exception:
394        return False
395
396    server_fp_val = _val(server_fp, "SHA256 Fingerprint")
397    return bool(server_fp_val) and server_fp_val == cert_fp_val
398
399
400# ---------------------------------------------------------------------------
401# Certificate details: SAN, validity period
402# ---------------------------------------------------------------------------
403
404def parse_san(text: str) -> list[str]:
405    """Return the DNS and IP entries of an ``openssl x509 -ext subjectAltName`` output.
406
407    ``DNS:web.tp, DNS:www.web.tp, IP Address:10.0.0.1`` -> ``['web.tp', 'www.web.tp', '10.0.0.1']``.
408    """
409    return [m.group(1).strip() for m in re.finditer(r'(?:DNS|IP Address)\s*:\s*([^,\s]+)', text or '')]
410
411
412def get_certificate_san(grade: Grade0, machine_name: str, cert_file: str,
413                        step: int = 1) -> list[str]:
414    """DNS/IP names of the subjectAltName extension of cert_file (``[]`` when absent or on error)."""
415    out, code = grade.test(
416        machine_name=machine_name,
417        command=f"openssl x509 -in {shlex.quote(cert_file)} -noout -ext subjectAltName",
418        step=step,
419        allow_error=True,
420    )
421    if code != 0:
422        return []
423    return parse_san(out)
424
425
426_MONTHS = {m: i + 1 for i, m in enumerate(('jan', 'feb', 'mar', 'apr', 'may', 'jun',
427                                            'jul', 'aug', 'sep', 'oct', 'nov', 'dec'))}
428_OPENSSL_DATE_RE = re.compile(r'([A-Za-z]{3})\s+(\d{1,2})\s+(\d{1,2}):(\d{2}):(\d{2})\s+(\d{4})')
429
430
431def parse_openssl_date(text: str) -> datetime | None:
432    """Parse an OpenSSL date such as ``Jan  1 00:00:00 2024 GMT`` (English month names,
433    whatever the locale of the grading process: ``strptime('%b')`` would not do)."""
434    m = _OPENSSL_DATE_RE.search(text or '')
435    if not m:
436        return None
437    month = _MONTHS.get(m.group(1).lower())
438    if month is None:
439        return None
440    try:
441        return datetime(int(m.group(6)), month, int(m.group(2)), int(m.group(3)), int(m.group(4)), int(m.group(5)))
442    except ValueError:
443        return None
444
445
446def certificate_validity_days(cert: dict | None) -> int | None:
447    """Validity period in days of a certificate dict (``not_before``/``not_after`` keys
448    as returned by :func:`eval_certificate`), ``None`` when the dates cannot be read."""
449    if not cert:
450        return None
451    start = parse_openssl_date(cert.get('not_before', ''))
452    end = parse_openssl_date(cert.get('not_after', ''))
453    if start is None or end is None:
454        return None
455    return round((end - start).total_seconds() / 86400)
456
457
458# ---------------------------------------------------------------------------
459# CRL
460# ---------------------------------------------------------------------------
461
462def normalize_serial(serial: str) -> str:
463    """Canonical form of a certificate serial number: upper-case hex, no colons, no
464    ``0x`` prefix, no leading zeros (``'0'`` for zero)."""
465    s = re.sub(r'[\s:]', '', (serial or '')).upper()
466    if s.startswith('0X'):
467        s = s[2:]
468    s = s.lstrip('0')
469    return s or '0'
470
471
472def parse_crl_text(text: str) -> list[str]:
473    """Serial numbers (normalised) listed as revoked in an ``openssl crl -text`` output."""
474    if 'Revoked Certificates' not in (text or ''):
475        return []
476    body = text.split('Revoked Certificates', 1)[1]
477    return [normalize_serial(m.group(1)) for m in re.finditer(r'Serial Number:\s*([0-9A-Fa-f:]+)', body)]
478
479
480def get_crl_revoked_serials(grade: Grade0, machine_name: str, crl_file: str,
481                            step: int = 1) -> list[str]:
482    """Revoked serial numbers of the CRL *crl_file* (``[]`` when unreadable)."""
483    out, code = grade.test(
484        machine_name=machine_name,
485        command=f"openssl crl -in {shlex.quote(crl_file)} -noout -text",
486        step=step,
487        allow_error=True,
488    )
489    if code != 0:
490        return []
491    return parse_crl_text(out)
492
493
494def eval_crl(grade: Grade0, machine_name: str, crl_file: str, ca_cert_file: str,
495             step: int = 1) -> bool:
496    """True when *crl_file* is a CRL whose signature verifies against *ca_cert_file*."""
497    out, code = grade.test(
498        machine_name=machine_name,
499        command=f"openssl crl -in {shlex.quote(crl_file)} -CAfile {shlex.quote(ca_cert_file)} -noout 2>&1",
500        step=step,
501        allow_error=True,
502    )
503    return code == 0 and 'verify failure' not in (out or '').lower()
504
505
506# ---------------------------------------------------------------------------
507# PKCS#12
508# ---------------------------------------------------------------------------
509
510_PEM_CERT_RE = re.compile(r'-----BEGIN CERTIFICATE-----.*?-----END CERTIFICATE-----', re.S)
511_CN_RE = re.compile(r'(?:^|[,/]\s*)CN\s*=\s*([^,/]+)')
512
513
514def cert_sha256_hex(pem: str) -> str:
515    """Lower-case hex SHA-256 of the DER form of a PEM certificate (``''`` if unparsable)."""
516    try:
517        return hashlib.sha256(ssl.PEM_cert_to_DER_cert(pem.strip())).hexdigest()
518    except Exception:
519        return ''
520
521
522def eval_pkcs12(grade: Grade0, machine_name: str, p12_file: str, password: str,
523                step: int = 1) -> dict | None:
524    """Open the PKCS#12 file *p12_file* with *password* on machine_name.
525
526    Returns ``None`` when the file cannot be opened (missing file or wrong password),
527    otherwise ``{'subject', 'common_name', 'fingerprint'}`` of the client certificate,
528    ``'ca_fingerprints'`` (SHA-256 fingerprints of the CA certificates bundled in the file)
529    and ``'has_key'`` (a private key is present).
530    """
531    q_file = shlex.quote(p12_file)
532    q_pass = shlex.quote(f'pass:{password}')
533    cert_out, cert_code = grade.test(
534        machine_name=machine_name,
535        command=f"openssl pkcs12 -in {q_file} -passin {q_pass} -nokeys -clcerts 2>/dev/null"
536                f" | openssl x509 -noout -subject -fingerprint -sha256",
537        step=step,
538        allow_error=True,
539    )
540    ca_out, _ = grade.test(
541        machine_name=machine_name,
542        command=f"openssl pkcs12 -in {q_file} -passin {q_pass} -nokeys -cacerts 2>/dev/null",
543        step=step,
544        allow_error=True,
545    )
546    _, key_code = grade.test(
547        machine_name=machine_name,
548        command=f"openssl pkcs12 -in {q_file} -passin {q_pass} -nocerts -nodes 2>/dev/null"
549                f" | openssl pkey -noout",
550        step=step,
551        allow_error=True,
552    )
553    if cert_code != 0 or 'subject=' not in (cert_out or ''):
554        return None
555    subject = _val(cert_out, 'subject')
556    cn_m = _CN_RE.search(subject)
557    ca_fps = []
558    for pem in _PEM_CERT_RE.findall(ca_out or ''):
559        try:
560            ca_fps.append(_cert_fingerprint_sha256(pem))
561        except Exception:
562            continue
563    return {
564        'subject': subject,
565        'common_name': cn_m.group(1).strip() if cn_m else '',
566        'fingerprint': _val(cert_out, 'SHA256 Fingerprint'),
567        'ca_fingerprints': ca_fps,
568        'has_key': key_code == 0,
569    }
570
571
572# ---------------------------------------------------------------------------
573# Live TLS servers
574# ---------------------------------------------------------------------------
575
576def get_tls_server_certificate(grade: Grade0, machine_name: str, server_ip: str,
577                               port: int = 443, servername: str | None = None,
578                               step: int = 1, starttls: str | None = None) -> dict | None:
579    """Certificate presented by the TLS server at *server_ip*:*port* (SNI *servername*),
580    as ``{'subject', 'issuer', 'common_name', 'fingerprint'}``; ``None`` when no
581    certificate could be read.  *starttls* names the protocol of an opportunistic TLS
582    server (``'smtp'``, ``'imap'``, ``'pop3'``...: ``openssl s_client -starttls``)."""
583    sni = f" -servername {shlex.quote(servername)}" if servername else ""
584    opportunistic = f" -starttls {shlex.quote(starttls)}" if starttls else ""
585    out, code = grade.test(
586        machine_name=machine_name,
587        command=f"openssl s_client -connect {_host_port(server_ip, int(port))}{sni}{opportunistic}"
588                f" </dev/null 2>/dev/null"
589                f" | openssl x509 -noout -subject -issuer -fingerprint -sha256",
590        step=step,
591        allow_error=True,
592    )
593    if code != 0 or 'subject=' not in (out or ''):
594        return None
595    subject = _val(out, 'subject')
596    cn_m = _CN_RE.search(subject)
597    return {
598        'subject': subject,
599        'issuer': _val(out, 'issuer'),
600        'common_name': cn_m.group(1).strip() if cn_m else '',
601        'fingerprint': _val(out, 'SHA256 Fingerprint'),
602    }
603
604
605_CURL_CODE_MARK = '===SRE_CODE '
606
607
608@dataclass
609class HttpResult:
610    """Result of :func:`https_get`: *code* is the HTTP status (``None`` when the request
611    failed before an answer, e.g. a certificate error), *headers* are lower-cased (the last value of a
612    repeated header), *all_headers* keeps every ``(name, value)`` in order (several ``Set-Cookie``)."""
613    code: int | None = None
614    headers: dict = field(default_factory=dict)
615    all_headers: list = field(default_factory=list)
616    body: str = ''
617    exit_code: int = 0
618
619    @property
620    def ok(self) -> bool:
621        return self.code is not None and 200 <= self.code < 300
622
623
624def parse_curl_output(out: str, exit_code: int = 0) -> HttpResult:
625    """Parse the output of ``curl -D - -o - -w '\\n===SRE_CODE %{http_code}'``."""
626    text = out or ''
627    code = None
628    if _CURL_CODE_MARK in text:
629        text, _, tail = text.rpartition(_CURL_CODE_MARK)
630        try:
631            code = int(tail.strip())
632        except ValueError:
633            code = None
634        if code == 0:
635            code = None
636    headers: dict = {}
637    all_headers: list = []
638    body = text
639    if text.startswith('HTTP/'):
640        m = re.search(r'\r?\n\r?\n', text)
641        head, body = (text[:m.start()], text[m.end():]) if m else (text, '')
642        for line in head.splitlines()[1:]:
643            if ':' in line:
644                name, value = line.split(':', 1)
645                headers[name.strip().lower()] = value.strip()
646                all_headers.append((name.strip(), value.strip()))
647    return HttpResult(code=code, headers=headers, all_headers=all_headers, body=body, exit_code=exit_code)
648
649
650def https_get(grade: Grade0, machine_name: str, url: str, server_ip: str,
651              port: int = 443, cacert: str | None = None, cert: str | None = None,
652              key: str | None = None, insecure: bool = False,
653              step: int = 1, timeout: int = 15) -> HttpResult:
654    """GET *url* from machine_name, connecting to *server_ip*:*port* for the URL's host
655    name (``curl --resolve``: no DNS needed, SNI and Host header come from the URL).
656
657    *cacert* / *cert* / *key* are paths on the machine (``--cacert``, ``--cert``, ``--key``);
658    *insecure* adds ``-k``. The status code, headers and body are returned in an
659    :class:`HttpResult` (``code`` is ``None`` when the TLS handshake or the connection failed).
660    """
661    host = urlparse(url).hostname or ''
662    opts = [f"--resolve {shlex.quote(f'{host}:{int(port)}:{_bracketed(server_ip)}')}"]
663    if insecure:
664        opts.append("-k")
665    if cacert:
666        opts.append(f"--cacert {shlex.quote(cacert)}")
667    if cert:
668        opts.append(f"--cert {shlex.quote(cert)}")
669    if key:
670        opts.append(f"--key {shlex.quote(key)}")
671    max_time = max(1, timeout - 3)
672    cmd = (f"curl -s -S --max-time {max_time} {' '.join(opts)} -D - -o - "
673           f"-w '\\n{_CURL_CODE_MARK}%{{http_code}}' {shlex.quote(url)} 2>&1")
674    out, code = grade.test(machine_name=machine_name, command=cmd, step=step,
675                           timeout=timeout, allow_error=True)
676    return parse_curl_output(out, code)
677
678
679# ---------------------------------------------------------------------------
680# Firefox (NSS) certificate store
681# ---------------------------------------------------------------------------
682
683def nss_hashes_command(home: str = '/root') -> str:
684    """One-line shell command printing the SHA-256 (hex) of every certificate stored in the
685    Firefox profiles under *home* (``cert9.db``, SQLite table ``nssPublic``, column ``a11`` =
686    CKA_VALUE). The databases are opened read-only and immutable so a running Firefox is
687    not disturbed. One hash per line; nothing when no profile exists.
688
689    The profiles are looked for in ``~/.mozilla/firefox`` and in ``~/.config/mozilla/firefox``:
690    Firefox 153 ESR (images 1.31) creates its profile in the second one when ``~/.mozilla`` does
691    not exist (seen live 2026-11-11: the first directory alone found nothing).
692
693    The script is base64-encoded into a ``python3 -c`` one-liner: the exetests runner only
694    supports single-line commands."""
695    script = (
696        "import glob, hashlib, sqlite3\n"
697        "seen = set()\n"
698        f"for db in glob.glob({home!r} + '/.mozilla/firefox*/*/cert9.db') "
699        f"+ glob.glob({home!r} + '/.config/mozilla/firefox*/*/cert9.db'):\n"
700        "    try:\n"
701        "        con = sqlite3.connect('file:' + db + '?mode=ro&immutable=1', uri=True)\n"
702        "        for (blob,) in con.execute('SELECT a11 FROM nssPublic WHERE a11 IS NOT NULL'):\n"
703        "            if isinstance(blob, bytes) and blob[:1] == b'\\x30':\n"
704        "                seen.add(hashlib.sha256(blob).hexdigest())\n"
705        "        con.close()\n"
706        "    except Exception:\n"
707        "        pass\n"
708        "print('\\n'.join(sorted(seen)))\n"
709    )
710    b64 = base64.b64encode(script.encode()).decode()
711    return f'python3 -c "import base64; exec(base64.b64decode(\'{b64}\').decode())"'
712
713
714def get_nss_certificate_hashes(grade: Grade0, machine_name: str, home: str = '/root',
715                               step: int = 1) -> set[str]:
716    """SHA-256 hashes of the certificates in the Firefox profiles of *home* on machine_name."""
717    out, code = grade.test(machine_name=machine_name, command=nss_hashes_command(home),
718                           step=step, allow_error=True)
719    if code != 0:
720        return set()
721    return {line.strip() for line in (out or '').splitlines() if re.fullmatch(r'[0-9a-f]{64}', line.strip())}
722
723
724def eval_firefox_certificate(grade: Grade0, machine_name: str, cert_pem: str,
725                             home: str = '/root', step: int = 1) -> bool:
726    """True when the PEM certificate *cert_pem* has been imported into a Firefox profile of
727    *home* on machine_name (any trust setting)."""
728    hashes = get_nss_certificate_hashes(grade, machine_name, home=home, step=step)
729    wanted = cert_sha256_hex(cert_pem)
730    return bool(wanted) and wanted in hashes
731
732
733# ---------------------------------------------------------------------------
734# step-ca
735# ---------------------------------------------------------------------------
736
737def generate_ca_pem(common_name: str, days: int = 3650, key_size: int = 2048) -> tuple[str, str]:
738    """``(certificate_pem, private_key_pem)`` of a fresh self-signed RSA certification authority
739    (``CA:TRUE``, keyCertSign / cRLSign, subject key identifier), for labs that hand a ready CA
740    to the students (the DNS lab's ``ca.tp`` signing the DoT / DoH certificate)."""
741    from datetime import datetime, timedelta, timezone
742    from cryptography import x509
743    from cryptography.hazmat.primitives import hashes, serialization
744    from cryptography.hazmat.primitives.asymmetric import rsa
745    from cryptography.x509.oid import NameOID
746    key = rsa.generate_private_key(public_exponent=65537, key_size=key_size)
747    name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, common_name)])
748    now = datetime.now(timezone.utc) - timedelta(days=1)
749    cert = (x509.CertificateBuilder().subject_name(name).issuer_name(name).public_key(key.public_key())
750            .serial_number(x509.random_serial_number()).not_valid_before(now)
751            .not_valid_after(now + timedelta(days=days))
752            .add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True)
753            .add_extension(x509.KeyUsage(digital_signature=True, content_commitment=False, key_encipherment=False,
754                                         data_encipherment=False, key_agreement=False, key_cert_sign=True,
755                                         crl_sign=True, encipher_only=False, decipher_only=False), critical=True)
756            .add_extension(x509.SubjectKeyIdentifier.from_public_key(key.public_key()), critical=False)
757            .sign(key, hashes.SHA256()))
758    cert_pem = cert.public_bytes(serialization.Encoding.PEM).decode()
759    key_pem = key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.TraditionalOpenSSL,
760                                serialization.NoEncryption()).decode()
761    return cert_pem, key_pem
762
763
764def generate_server_cert_pem(ca_cert_pem: str, ca_key_pem: str, hostname: str, days: int = 365,
765                             key_size: int = 2048, extra_names: tuple[str, ...] = ()) -> tuple[str, str]:
766    """``(certificate_pem, private_key_pem)`` of a fresh RSA server certificate for *hostname*
767    (CN and ``subjectAltName`` dNSName, plus *extra_names*), signed by the CA of
768    :func:`generate_ca_pem`: what a reference solution installs on a TLS server (the LDAP lab's
769    ``ldap1.corp.lan``) without running ``openssl`` in the container."""
770    from datetime import datetime, timedelta, timezone
771    from cryptography import x509
772    from cryptography.hazmat.primitives import hashes, serialization
773    from cryptography.hazmat.primitives.asymmetric import rsa
774    from cryptography.x509.oid import NameOID, ExtendedKeyUsageOID
775    ca_cert = x509.load_pem_x509_certificate(ca_cert_pem.encode())
776    ca_key = serialization.load_pem_private_key(ca_key_pem.encode(), password=None)
777    key = rsa.generate_private_key(public_exponent=65537, key_size=key_size)
778    now = datetime.now(timezone.utc) - timedelta(days=1)
779    names = [x509.DNSName(hostname)] + [x509.DNSName(n) for n in extra_names]
780    cert = (x509.CertificateBuilder()
781            .subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, hostname)]))
782            .issuer_name(ca_cert.subject).public_key(key.public_key())
783            .serial_number(x509.random_serial_number()).not_valid_before(now)
784            .not_valid_after(now + timedelta(days=days))
785            .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
786            .add_extension(x509.KeyUsage(digital_signature=True, content_commitment=False, key_encipherment=True,
787                                         data_encipherment=False, key_agreement=False, key_cert_sign=False,
788                                         crl_sign=False, encipher_only=False, decipher_only=False), critical=True)
789            .add_extension(x509.ExtendedKeyUsage([ExtendedKeyUsageOID.SERVER_AUTH]), critical=False)
790            .add_extension(x509.SubjectAlternativeName(names), critical=False)
791            .add_extension(x509.SubjectKeyIdentifier.from_public_key(key.public_key()), critical=False)
792            .add_extension(x509.AuthorityKeyIdentifier.from_issuer_public_key(ca_key.public_key()), critical=False)
793            .sign(ca_key, hashes.SHA256()))
794    cert_pem = cert.public_bytes(serialization.Encoding.PEM).decode()
795    key_pem = key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.TraditionalOpenSSL,
796                                serialization.NoEncryption()).decode()
797    return cert_pem, key_pem
798
799
800def parse_stepca_config(text: str) -> dict | None:
801    """Summary of a step-ca ``ca.json``: ``{'address', 'dns_names', 'root', 'provisioners':
802    [{'name', 'type'}, ...]}``; ``None`` when the text is not a JSON object."""
803    try:
804        cfg = json.loads(text or '')
805    except (ValueError, TypeError):
806        return None
807    if not isinstance(cfg, dict):
808        return None
809    provisioners = (cfg.get('authority') or {}).get('provisioners') or []
810    return {
811        'address': cfg.get('address', ''),
812        'dns_names': list(cfg.get('dnsNames') or []),
813        'root': cfg.get('root', ''),
814        'provisioners': [{'name': p.get('name', ''), 'type': p.get('type', '')}
815                         for p in provisioners if isinstance(p, dict)],
816    }
817
818
819def get_stepca_config(grade: Grade0, machine_name: str,
820                      config_file: str = '/home/ca/.step/config/ca.json',
821                      step: int = 1) -> dict | None:
822    """Parsed step-ca configuration of machine_name (see :func:`parse_stepca_config`)."""
823    out, code = grade.test(machine_name=machine_name, command=f"cat {shlex.quote(config_file)}",
824                           step=step, allow_error=True)
825    if code != 0:
826        return None
827    return parse_stepca_config(out)