diff --git a/CHANGELOG.md b/CHANGELOG.md index 0cc1f4e..f4d6d99 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,52 @@ Version numbers follow [Semantic Versioning](https://semver.org/spec/v2.0.0.html --- +## [0.7.0] — 2026-08-17 + +### Added +- **PQC-02 — certificate algorithm tracking**: new `certificate_key` check on + every TLS/STARTTLS assessment. The leaf certificate is retrieved with a + second `openssl s_client` probe (the primary probe runs with `-brief`, + which suppresses certificate output) and parsed with pyca/cryptography. + RSA keys below 3072 bits (CNSA 2.0 / BSI TR-02102-2), EC curves outside + P-256/P-384/P-521/Brainpool ≥ 256 (NIST SP 800-186), and DSA keys + (withdrawn, FIPS 186-5) are reported as `FAIL`; Ed25519/Ed448 pass. If the + certificate cannot be retrieved or parsed the check is `INFO`, never + `ERROR` — a failed best-effort fetch must not flip the CLI to exit code 2 + or create a platform finding. +- **PQC-03 — SSH host-key algorithm check**: new `host_key_algorithms` check + on every SSH assessment. The `server_host_key_algorithms` name-list is + parsed from the same KEXINIT packet already read for the KEX check + (RFC 4253 §7.1) — no extra connection. Advertising `ssh-dss` (DSA, + withdrawn by FIPS 186-5) or `ssh-rsa` (SHA-1 signatures, superseded by + rsa-sha2-256/512 per RFC 8332), including their `*-cert-v01@openssh.com` + variants, is reported as `FAIL`. +- `tls_utils.CertificateInfo` dataclass and `tls_utils.fetch_certificate()`; + `TLSProbeResult` gained optional `certificate` and + `ssh_host_key_algorithms` fields. +- `constants`: `RSA_MIN_KEY_SIZE`, `EC_APPROVED_CURVES`, `EC_CURVE_DISPLAY`, + `DEPRECATED_SSH_HOST_KEY_ALGORITHMS`. +- New runtime dependency: `cryptography>=42` (X.509 parsing). Chosen over + scraping `openssl x509 -text` output, which is fragile across OpenSSL + versions and locales. + +### Changed +- `verdict.build_checks()` now accepts keyword-only `certificate` and + `ssh_host_key_algorithms` arguments and returns **three** checks instead of + two for both TLS and SSH probes. +- Successful TLS assessments now open one additional connection (the + certificate fetch). SSH assessments are unchanged (single connection). + +### Unchanged by design (backend contract) +- The SAFE/UNSAFE verdict still tracks PQC key-exchange readiness (PQC-01) + only — `certificate_key` / `host_key_algorithms` failures do **not** flip + the verdict or the CLI exit code. `Status` and `Verdict` enums gained no + new members; existing check names are untouched. The platform's + severity-vocabulary mapping (IDR-018) keeps working without backend + changes; the two new check names simply appear as additional findings. + +--- + ## [0.6.4] — 2026-08-12 ### Added @@ -328,7 +374,8 @@ Version numbers follow [Semantic Versioning](https://semver.org/spec/v2.0.0.html --- -[Unreleased]: https://github.com/NC3-TestingPlatform/quantumvalidator/compare/v0.6.4...HEAD +[Unreleased]: https://github.com/NC3-TestingPlatform/quantumvalidator/compare/v0.7.0...HEAD +[0.7.0]: https://github.com/NC3-TestingPlatform/quantumvalidator/compare/v0.6.4...v0.7.0 [0.6.4]: https://github.com/NC3-TestingPlatform/quantumvalidator/compare/v0.6.3...v0.6.4 [0.6.3]: https://github.com/NC3-TestingPlatform/quantumvalidator/compare/v0.6.2...v0.6.3 [0.6.2]: https://github.com/NC3-TestingPlatform/quantumvalidator/compare/v0.6.1...v0.6.2 diff --git a/README.md b/README.md index 998b111..ae53a59 100644 --- a/README.md +++ b/README.md @@ -6,13 +6,15 @@ FTP/STARTTLS, LMTP/STARTTLS, NNTP/STARTTLS, ManageSieve/STARTTLS, and SSH endpoints to detect whether a PQC hybrid key exchange (ML-KEM) was negotiated, returning a binary **SAFE** / **UNSAFE** verdict aligned with NSA CNSA 2.0, BSI TR-02102-2 (TLS), and BSI TR-02102-4 (SSH). +It also tracks classical algorithm hygiene: certificate key algorithms (RSA size, EC curve) +for TLS services, and deprecated host-key algorithms (DSA, RSA-SHA1) for SSH services. ```bash $ quantumvalidator check cloudflare.com ``` ![Python](https://img.shields.io/badge/python-%3E%3D3.11-blue) -![Tests](https://img.shields.io/badge/tests-247%20passing-brightgreen) +![Tests](https://img.shields.io/badge/tests-302%20passing-brightgreen) ![Coverage](https://img.shields.io/badge/coverage-100%25-brightgreen) ![License](https://img.shields.io/badge/license-GPLv3-lightgrey) @@ -61,6 +63,22 @@ The server must advertise at least one of these algorithms in its KEXINIT to rec verdict. `sntrup761x25519-sha512@openssh.com` (NTRU) is intentionally excluded — it uses a non-NIST-selected algorithm. +### Classical algorithm hygiene checks + +Alongside PQC key-exchange readiness, two additional checks flag weak classical algorithms. +Their failures do **not** change the SAFE/UNSAFE verdict — they surface as `FAIL` entries in +the check table: + +| Check | Protocol | FAIL condition | Standard | +|---|---|---|---| +| `certificate_key` | TLS | RSA < 3072 bits, EC curve outside P-256/P-384/P-521/Brainpool ≥ 256, or DSA | CNSA 2.0, BSI TR-02102-2, NIST SP 800-186, FIPS 186-5 | +| `host_key_algorithms` | SSH | `ssh-dss` (DSA, withdrawn) or `ssh-rsa` (SHA-1 signatures) advertised | RFC 8332, FIPS 186-5, BSI TR-02102-4 | + +The certificate is retrieved with a second `openssl s_client` probe (the primary probe runs +with `-brief`, which suppresses certificate output) and parsed with +[pyca/cryptography](https://cryptography.io/). The SSH host-key list is read from the same +KEXINIT packet as the KEX algorithms — no extra connection. + --- ## Standards @@ -82,7 +100,7 @@ non-NIST-selected algorithm. - **OpenSSL** ≥ 3.5 binary on PATH — required for native PQC hybrid group negotiation - Debian/Ubuntu: `apt install openssl` - macOS: `brew install openssl` -- `rich` ≥ 13.7 and `typer` ≥ 0.12 (installed automatically via pip) +- `cryptography` ≥ 42, `rich` ≥ 13.7, and `typer` ≥ 0.12 (installed automatically via pip) --- @@ -235,7 +253,8 @@ if report.is_safe: The verdict is binary by design — classical certificate signatures are not yet a practical threat (no public CA issues ML-DSA/SLH-DSA certificates), so the signal focuses solely on -key exchange. +key exchange. The `certificate_key` and `host_key_algorithms` checks report classical +algorithm weaknesses independently, without changing the verdict. --- @@ -297,7 +316,7 @@ pytest tests/test_tls_utils.py pytest tests/test_assessor.py::TestAssessHttps -v ``` -The test suite has **247 tests** and maintains **100% statement coverage**. +The test suite has **302 tests** and maintains **100% statement coverage**. All network I/O (`openssl s_client` subprocess) is mocked at the `probe_tls` boundary — no test touches a real server or the internet. diff --git a/pyproject.toml b/pyproject.toml index 50410b0..dfe5961 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "quantumvalidator" -version = "0.6.4" +version = "0.7.0" description = "Quantum-safe cryptography validator — TLS, STARTTLS, and SSH post-quantum readiness assessment" readme = "README.md" requires-python = ">=3.11" @@ -41,6 +41,7 @@ classifiers = [ "Typing :: Typed", ] dependencies = [ + "cryptography>=42", "rich>=13.7", "typer>=0.12", ] diff --git a/quantumvalidator/__init__.py b/quantumvalidator/__init__.py index 6eb93fe..a92a0fe 100644 --- a/quantumvalidator/__init__.py +++ b/quantumvalidator/__init__.py @@ -8,7 +8,7 @@ try: __version__ = version("quantumvalidator") except PackageNotFoundError: # pragma: no cover - __version__ = "0.6.4" + __version__ = "0.7.0" _logging.getLogger("quantumvalidator").addHandler(_logging.NullHandler()) del _logging diff --git a/quantumvalidator/assessor.py b/quantumvalidator/assessor.py index 11cde49..18f6257 100644 --- a/quantumvalidator/assessor.py +++ b/quantumvalidator/assessor.py @@ -38,6 +38,11 @@ def assess( The probe auto-detects STARTTLS mode (smtp/imap/pop3/ftp/lmtp/nntp/sieve) via banner fingerprinting, and SSH via ``SSH-2.0-`` banner. + Besides PQC key-exchange readiness, the report includes classical + algorithm hygiene checks: ``certificate_key`` (RSA size / EC curve of the + TLS leaf certificate) and ``host_key_algorithms`` (deprecated SSH + host-key algorithms). Their failures do not affect the verdict. + :param target: Hostname or IP address to probe. :param port: TCP port override; defaults to 443. :param timeout: Connection timeout in seconds. @@ -68,7 +73,12 @@ def assess( ] verdict = Verdict.UNSAFE else: - checks = build_checks(result.tls_version, result.negotiated_group) + checks = build_checks( + result.tls_version, + result.negotiated_group, + certificate=result.certificate, + ssh_host_key_algorithms=result.ssh_host_key_algorithms, + ) verdict = determine_verdict(result.tls_version, result.negotiated_group) report = QuantumReport( diff --git a/quantumvalidator/cli.py b/quantumvalidator/cli.py index 161aef9..ef4f496 100644 --- a/quantumvalidator/cli.py +++ b/quantumvalidator/cli.py @@ -35,7 +35,9 @@ "Auto-detects STARTTLS protocols (SMTP, IMAP, POP3, FTP, LMTP, NNTP, Sieve) " "via banner fingerprinting, and SSH via SSH-2.0- banner. " "Checks whether the service negotiates a PQC hybrid key exchange group (ML-KEM) " - "as required by CNSA 2.0 and BSI TR-02102-2/TR-02102-4." + "as required by CNSA 2.0 and BSI TR-02102-2/TR-02102-4. " + "Also tracks certificate key algorithms (RSA size, EC curve) and " + "deprecated SSH host-key algorithms (DSA, RSA-SHA1)." ), add_completion=False, ) diff --git a/quantumvalidator/constants.py b/quantumvalidator/constants.py index 945b161..c410207 100644 --- a/quantumvalidator/constants.py +++ b/quantumvalidator/constants.py @@ -92,6 +92,47 @@ class GroupInfo: name for name, g in SSH_PQC_GROUPS.items() if g.safe ) +# Certificate-key policy (PQC-02). +# RSA below 3072 bits is under the CNSA 2.0 floor and below the +# BSI TR-02102-2 recommendation of >= 3000 bits. +RSA_MIN_KEY_SIZE: int = 3072 + +# Curves approved for TLS server certificates: NIST P-256/P-384/P-521 +# (NIST SP 800-186) plus the >= 256-bit Brainpool curves accepted by +# BSI TR-02102-2. Keys are the pyca/cryptography ``curve.name`` values. +EC_APPROVED_CURVES: frozenset[str] = frozenset( + { + "secp256r1", + "secp384r1", + "secp521r1", + "brainpoolP256r1", + "brainpoolP384r1", + "brainpoolP512r1", + } +) + +# pyca/cryptography curve.name → common display name. Curves absent from +# this map are displayed under their SECG name (e.g. 'secp256k1'). +EC_CURVE_DISPLAY: dict[str, str] = { + "secp192r1": "P-192", + "secp224r1": "P-224", + "secp256r1": "P-256", + "secp384r1": "P-384", + "secp521r1": "P-521", +} + +# SSH host-key algorithms rejected by PQC-03: ssh-dss is DSA (withdrawn by +# NIST FIPS 186-5); ssh-rsa signs with SHA-1 (superseded by rsa-sha2-256/512, +# RFC 8332). The *-cert-v01 variants sign with the same primitives. +DEPRECATED_SSH_HOST_KEY_ALGORITHMS: frozenset[str] = frozenset( + { + "ssh-dss", + "ssh-rsa", + "ssh-dss-cert-v01@openssh.com", + "ssh-rsa-cert-v01@openssh.com", + } +) + DEFAULT_TIMEOUT: float = 10.0 DEFAULT_PORT_HTTPS: int = 443 diff --git a/quantumvalidator/tls_utils.py b/quantumvalidator/tls_utils.py index c7c1d39..81b194d 100644 --- a/quantumvalidator/tls_utils.py +++ b/quantumvalidator/tls_utils.py @@ -15,6 +15,9 @@ import subprocess from dataclasses import dataclass, replace +from cryptography import x509 +from cryptography.hazmat.primitives.asymmetric import dsa, ec, ed448, ed25519, rsa + from quantumvalidator.constants import ( OPENSSL_BINARY, PROBE_GROUPS, @@ -52,6 +55,18 @@ def _validate_target(host: str, port: int) -> None: raise ValueError(f"Invalid hostname: {host!r}") +@dataclass(frozen=True) +class CertificateInfo: + """Public-key algorithm details of a server's leaf certificate.""" + + key_type: str + """Key algorithm family: ``'RSA'``/``'EC'``/``'Ed25519'``/``'Ed448'``/``'DSA'``/``'unknown'``.""" + key_size: int | None + """Key size in bits (modulus for RSA, curve size for EC), or None.""" + curve: str | None + """pyca/cryptography curve name (e.g. ``'secp256r1'``) for EC keys, else None.""" + + @dataclass class TLSProbeResult: """Raw output of a single openssl s_client probe.""" @@ -66,6 +81,10 @@ class TLSProbeResult: """Error message if the probe failed, None on success.""" detected_starttls: str | None = None """STARTTLS mode detected from server banner (e.g. ``'smtp'``/``'ftp'``/``'nntp'``/``'ssh'``), or ``None``.""" + certificate: CertificateInfo | None = None + """Leaf-certificate key details (TLS probes only), or ``None`` if unavailable.""" + ssh_host_key_algorithms: list[str] | None = None + """server_host_key_algorithms from the SSH KEXINIT, or ``None`` (non-SSH or parse failure).""" @property def ok(self) -> bool: @@ -164,6 +183,86 @@ def probe_raw( return None +_PEM_CERT_RE: re.Pattern[str] = re.compile( + r"-----BEGIN CERTIFICATE-----.*?-----END CERTIFICATE-----", + re.DOTALL, +) + + +def _extract_pem_cert(output: str) -> str | None: + """Extract the first PEM certificate block from openssl s_client output. + + Without ``-showcerts``, s_client prints only the leaf certificate, so the + first block is the server certificate. + + :param output: Combined stdout+stderr from ``openssl s_client``. + :returns: PEM certificate string, or ``None`` if no block is present. + :rtype: str | None + """ + m = _PEM_CERT_RE.search(output) + return m.group(0) if m else None + + +def _parse_certificate(pem: str) -> CertificateInfo | None: + """Parse a PEM certificate and return its public-key algorithm details. + + :param pem: PEM-encoded X.509 certificate. + :returns: Key details, or ``None`` if the PEM cannot be parsed. + :rtype: CertificateInfo | None + """ + try: + cert = x509.load_pem_x509_certificate(pem.encode("ascii")) + except ValueError as exc: + logger.debug("Certificate parse failed: %s", exc) + return None + key = cert.public_key() + if isinstance(key, rsa.RSAPublicKey): + return CertificateInfo(key_type="RSA", key_size=key.key_size, curve=None) + if isinstance(key, ec.EllipticCurvePublicKey): + return CertificateInfo( + key_type="EC", key_size=key.curve.key_size, curve=key.curve.name + ) + if isinstance(key, ed25519.Ed25519PublicKey): + return CertificateInfo(key_type="Ed25519", key_size=256, curve=None) + if isinstance(key, ed448.Ed448PublicKey): + return CertificateInfo(key_type="Ed448", key_size=456, curve=None) + if isinstance(key, dsa.DSAPublicKey): + return CertificateInfo(key_type="DSA", key_size=key.key_size, curve=None) + return CertificateInfo(key_type="unknown", key_size=None, curve=None) + + +def fetch_certificate( + host: str, + port: int, + *, + starttls: str | None = None, + timeout: float = 10.0, +) -> CertificateInfo | None: + """Fetch the leaf certificate of *host*:*port* and return its key details. + + Runs a second ``openssl s_client`` probe without ``-brief`` (via + :func:`probe_raw`) because ``-brief`` suppresses certificate output, then + parses the PEM block with pyca/cryptography. + + :param host: Hostname or IP to connect to. + :param port: TCP port. + :param starttls: openssl ``-starttls`` mode, or ``None`` for raw TLS. + :param timeout: Connection timeout in seconds. + :returns: Certificate key details, or ``None`` if the certificate could + not be retrieved or parsed. + :rtype: CertificateInfo | None + """ + output = probe_raw(host, port, starttls=starttls, timeout=timeout) + if output is None: + logger.warning("Certificate fetch failed for %s:%d", host, port) + return None + pem = _extract_pem_cert(output) + if pem is None: + logger.warning("No certificate in s_client output for %s:%d", host, port) + return None + return _parse_certificate(pem) + + def probe_tls( host: str, port: int, @@ -203,17 +302,23 @@ def probe_tls( return _probe_ssh(host, port, timeout) if detected == "ftp": - return _probe_ftp(host, port, timeout) - - if detected: - logger.info( - "Banner detected '%s' for %s:%d — probing with -starttls %s", - detected, host, port, detected, + result = _probe_ftp(host, port, timeout) + else: + if detected: + logger.info( + "Banner detected '%s' for %s:%d — probing with -starttls %s", + detected, host, port, detected, + ) + result = _run_openssl(host, port, starttls=detected, timeout=timeout) + if detected: + result.detected_starttls = detected + + # PQC-02: fetch the leaf certificate with a second, non--brief probe. + # Failure here never fails the assessment — certificate stays None. + if result.ok: + result.certificate = fetch_certificate( + host, port, starttls=result.detected_starttls, timeout=timeout ) - - result = _run_openssl(host, port, starttls=detected, timeout=timeout) - if detected: - result.detected_starttls = detected return result @@ -366,25 +471,38 @@ def _read_ssh_packet(sock: socket.socket, prepend: bytes = b"") -> bytes: return payload[:packet_length] -def _parse_ssh_kexinit(packet: bytes) -> list[str]: - """Parse an SSH_MSG_KEXINIT packet; return the kex_algorithms name-list. +def _parse_ssh_kexinit(packet: bytes) -> tuple[list[str], list[str]]: + """Parse an SSH_MSG_KEXINIT packet; return two name-lists. Layout (RFC 4253 §7.1): - padding_length(1) msg_type(1) cookie(16) kex_algorithms(name-list) … + padding_length(1) msg_type(1) cookie(16) kex_algorithms(name-list) + server_host_key_algorithms(name-list) … :param packet: Raw packet bytes from ``_read_ssh_packet``. - :returns: List of KEX algorithm names, or ``[]`` on parse failure. - :rtype: list[str] + :returns: ``(kex_algorithms, server_host_key_algorithms)``; either list is + ``[]`` when its name-list cannot be parsed. + :rtype: tuple[list[str], list[str]] """ # Minimum: padding_len(1) + msg_type(1) + cookie(16) + namelist_len(4) = 22 if len(packet) < 22 or packet[1] != 20: # 20 = SSH_MSG_KEXINIT - return [] + return [], [] offset = 18 # 1(padding_len) + 1(msg_type) + 16(cookie) name_list_length = int.from_bytes(packet[offset: offset + 4], "big") if len(packet) < offset + 4 + name_list_length: - return [] + return [], [] + raw = packet[offset + 4: offset + 4 + name_list_length] + kex_algorithms = raw.decode("ascii", errors="replace").split(",") + + # server_host_key_algorithms is the next name-list (PQC-03). + offset += 4 + name_list_length + if len(packet) < offset + 4: + return kex_algorithms, [] + name_list_length = int.from_bytes(packet[offset: offset + 4], "big") + if len(packet) < offset + 4 + name_list_length: + return kex_algorithms, [] raw = packet[offset + 4: offset + 4 + name_list_length] - return raw.decode("ascii", errors="replace").split(",") + host_key_algorithms = raw.decode("ascii", errors="replace").split(",") + return kex_algorithms, host_key_algorithms def _probe_ssh(host: str, port: int, timeout: float) -> TLSProbeResult: @@ -431,7 +549,7 @@ def _probe_ssh(host: str, port: int, timeout: float) -> TLSProbeResult: sock.sendall(b"SSH-2.0-quantumvalidator_0.1\r\n") packet = _read_ssh_packet(sock, leftover) - kex_algorithms = _parse_ssh_kexinit(packet) + kex_algorithms, host_key_algorithms = _parse_ssh_kexinit(packet) except (OSError, socket.timeout) as exc: logger.warning("SSH probe failed for %s:%d: %s", host, port, exc) @@ -451,8 +569,8 @@ def _probe_ssh(host: str, port: int, timeout: float) -> TLSProbeResult: ) logger.info( - "SSH KEXINIT from %s:%d — %d algorithms, first: %s", - host, port, len(kex_algorithms), kex_algorithms[0], + "SSH KEXINIT from %s:%d — %d algorithms, first: %s; %d host-key algorithms", + host, port, len(kex_algorithms), kex_algorithms[0], len(host_key_algorithms), ) safe_kex = next( @@ -463,6 +581,7 @@ def _probe_ssh(host: str, port: int, timeout: float) -> TLSProbeResult: tls_version="SSHv2", negotiated_group=safe_kex or kex_algorithms[0], detected_starttls="ssh", + ssh_host_key_algorithms=[a for a in host_key_algorithms if a] or None, ) diff --git a/quantumvalidator/verdict.py b/quantumvalidator/verdict.py index 4a4a82b..43ad519 100644 --- a/quantumvalidator/verdict.py +++ b/quantumvalidator/verdict.py @@ -10,9 +10,23 @@ from __future__ import annotations -from quantumvalidator.constants import PQC_GROUPS, SAFE_GROUPS, SSH_PQC_GROUPS, SSH_SAFE_GROUPS +from typing import TYPE_CHECKING + +from quantumvalidator.constants import ( + DEPRECATED_SSH_HOST_KEY_ALGORITHMS, + EC_APPROVED_CURVES, + EC_CURVE_DISPLAY, + PQC_GROUPS, + RSA_MIN_KEY_SIZE, + SAFE_GROUPS, + SSH_PQC_GROUPS, + SSH_SAFE_GROUPS, +) from quantumvalidator.models import CheckResult, Status, Verdict +if TYPE_CHECKING: + from quantumvalidator.tls_utils import CertificateInfo + def determine_verdict( tls_version: str | None, @@ -44,19 +58,30 @@ def determine_verdict( def build_checks( tls_version: str | None, negotiated_group: str | None, + *, + certificate: CertificateInfo | None = None, + ssh_host_key_algorithms: list[str] | None = None, ) -> list[CheckResult]: """Build the ordered list of CheckResult objects for a probe outcome. - Produces two checks in order: version check, then key-exchange check. - Dispatches to SSH-specific logic when ``tls_version == "SSHv2"``. + For TLS, produces three checks: version, key exchange, certificate key + (PQC-02). For SSH (``tls_version == "SSHv2"``), produces three checks: + version, KEX algorithm, host-key algorithms (PQC-03). + + Certificate and host-key failures do **not** affect the SAFE/UNSAFE + verdict, which tracks PQC key-exchange readiness only — they surface as + FAIL checks with their own weight downstream. :param tls_version: Protocol version string (``'TLSv1.3'``, ``'SSHv2'``, or None). :param negotiated_group: Negotiated/best KEX group name, or None. - :returns: List of two CheckResult items. + :param certificate: Leaf-certificate key details, or None if unavailable. + :param ssh_host_key_algorithms: server_host_key_algorithms from the SSH + KEXINIT, or None if unavailable. + :returns: List of three CheckResult items. :rtype: list[CheckResult] """ if tls_version == "SSHv2": - return _build_ssh_checks(negotiated_group) + return _build_ssh_checks(negotiated_group, ssh_host_key_algorithms) checks: list[CheckResult] = [] @@ -103,14 +128,171 @@ def build_checks( standard="CNSA 2.0, BSI TR-02102-2", )) + # Check 3: Certificate key algorithm (PQC-02) + checks.append(_build_certificate_check(certificate)) + return checks -def _build_ssh_checks(negotiated_group: str | None) -> list[CheckResult]: - """Build two CheckResult items for an SSH probe outcome. +def _build_certificate_check(certificate: CertificateInfo | None) -> CheckResult: + """Build the PQC-02 certificate-key CheckResult. + + Flags RSA below :data:`RSA_MIN_KEY_SIZE` bits, EC curves outside + :data:`EC_APPROVED_CURVES`, and DSA keys. A missing certificate yields + INFO, never ERROR — the fetch is a best-effort second probe and must not + fail an otherwise successful assessment. + + :param certificate: Leaf-certificate key details, or None if unavailable. + :returns: CheckResult named ``certificate_key``. + :rtype: CheckResult + """ + if certificate is None: + return CheckResult( + name="certificate_key", + status=Status.INFO, + value=None, + reason="Certificate could not be retrieved; key algorithm not assessed.", + standard=None, + ) + + if certificate.key_type == "RSA": + value = f"RSA-{certificate.key_size}" + if certificate.key_size is not None and certificate.key_size >= RSA_MIN_KEY_SIZE: + return CheckResult( + name="certificate_key", + status=Status.PASS, + value=value, + reason=( + f"RSA key meets the {RSA_MIN_KEY_SIZE}-bit classical minimum " + "(all pre-PQC certificate keys remain quantum-vulnerable)." + ), + standard="CNSA 2.0, BSI TR-02102-2", + ) + return CheckResult( + name="certificate_key", + status=Status.FAIL, + value=value, + reason=( + f"RSA key below the {RSA_MIN_KEY_SIZE}-bit classical minimum. " + f"Reissue the certificate with RSA >= {RSA_MIN_KEY_SIZE} or an " + "approved EC curve (P-256/P-384/P-521)." + ), + standard="CNSA 2.0, BSI TR-02102-2", + ) + + if certificate.key_type == "EC": + curve = certificate.curve or "unknown" + display = EC_CURVE_DISPLAY.get(curve, curve) + if curve in EC_APPROVED_CURVES: + return CheckResult( + name="certificate_key", + status=Status.PASS, + value=display, + reason=( + "EC curve is approved for certificates " + "(all pre-PQC certificate keys remain quantum-vulnerable)." + ), + standard="NIST SP 800-186, BSI TR-02102-2", + ) + return CheckResult( + name="certificate_key", + status=Status.FAIL, + value=display, + reason=( + f"Deprecated or unapproved EC curve {display}. " + "Reissue the certificate on P-256, P-384, or P-521." + ), + standard="NIST SP 800-186, BSI TR-02102-2", + ) + + if certificate.key_type in ("Ed25519", "Ed448"): + return CheckResult( + name="certificate_key", + status=Status.PASS, + value=certificate.key_type, + reason=( + "Modern EdDSA certificate key " + "(all pre-PQC certificate keys remain quantum-vulnerable)." + ), + standard="RFC 8032, RFC 8410", + ) + + if certificate.key_type == "DSA": + return CheckResult( + name="certificate_key", + status=Status.FAIL, + value=f"DSA-{certificate.key_size}", + reason=( + "DSA is withdrawn for digital signatures. Reissue the " + f"certificate with RSA >= {RSA_MIN_KEY_SIZE} or an approved EC curve." + ), + standard="NIST FIPS 186-5", + ) + + return CheckResult( + name="certificate_key", + status=Status.INFO, + value=certificate.key_type, + reason="Unrecognised certificate key algorithm; not assessed.", + standard=None, + ) + + +def _build_host_key_check(host_key_algorithms: list[str] | None) -> CheckResult: + """Build the PQC-03 SSH host-key CheckResult. + + Flags deprecated host-key algorithms from + :data:`DEPRECATED_SSH_HOST_KEY_ALGORITHMS` (DSA and RSA-with-SHA-1). + + :param host_key_algorithms: server_host_key_algorithms name-list, or None. + :returns: CheckResult named ``host_key_algorithms``. + :rtype: CheckResult + """ + if not host_key_algorithms: + return CheckResult( + name="host_key_algorithms", + status=Status.INFO, + value=None, + reason="Host-key algorithm list unavailable; not assessed.", + standard=None, + ) + + deprecated = [ + alg for alg in host_key_algorithms + if alg in DEPRECATED_SSH_HOST_KEY_ALGORITHMS + ] + if deprecated: + return CheckResult( + name="host_key_algorithms", + status=Status.FAIL, + value=",".join(deprecated), + reason=( + "Deprecated host-key algorithms advertised: ssh-dss is DSA " + "(withdrawn by FIPS 186-5); ssh-rsa signs with SHA-1 — " + "use rsa-sha2-256/512 instead." + ), + standard="RFC 8332, NIST FIPS 186-5, BSI TR-02102-4", + ) + + return CheckResult( + name="host_key_algorithms", + status=Status.PASS, + value=",".join(host_key_algorithms), + reason="No deprecated host-key algorithms (DSA, RSA-SHA1) advertised.", + standard="RFC 8332, NIST FIPS 186-5, BSI TR-02102-4", + ) + + +def _build_ssh_checks( + negotiated_group: str | None, + host_key_algorithms: list[str] | None = None, +) -> list[CheckResult]: + """Build three CheckResult items for an SSH probe outcome. :param negotiated_group: Best PQC KEX algorithm found, or first classical one. - :returns: List of two CheckResult items: ssh_version then kex_algorithm. + :param host_key_algorithms: server_host_key_algorithms name-list, or None. + :returns: List of three CheckResult items: ssh_version, kex_algorithm, + host_key_algorithms. :rtype: list[CheckResult] """ checks: list[CheckResult] = [] @@ -145,4 +327,7 @@ def _build_ssh_checks(negotiated_group: str | None) -> list[CheckResult]: standard="NIST FIPS 203", )) + # Check 3: Host-key algorithms (PQC-03) + checks.append(_build_host_key_check(host_key_algorithms)) + return checks diff --git a/tests/test_assessor.py b/tests/test_assessor.py index 7150b8c..35167b3 100644 --- a/tests/test_assessor.py +++ b/tests/test_assessor.py @@ -13,6 +13,7 @@ ) from quantumvalidator.assessor import assess from quantumvalidator.models import Status, Verdict +from quantumvalidator.tls_utils import CertificateInfo class TestAssess: @@ -51,14 +52,46 @@ def test_error_probe_single_error_check(self, monkeypatch): assert report.checks[0].name == "connection" assert "Connection refused" in report.checks[0].reason - def test_two_pass_checks_on_success(self, monkeypatch): + def test_three_checks_on_success(self, monkeypatch): monkeypatch.setattr( "quantumvalidator.assessor.check_tls", lambda *a, **kw: SAFE_PROBE ) report = assess("example.com") - assert len(report.checks) == 2 - assert [c.name for c in report.checks] == ["tls_version", "key_exchange"] - assert all(c.status == Status.PASS for c in report.checks) + assert len(report.checks) == 3 + assert [c.name for c in report.checks] == [ + "tls_version", "key_exchange", "certificate_key", + ] + assert all(c.status == Status.PASS for c in report.checks[:2]) + + def test_certificate_passed_to_checks(self, monkeypatch): + probe = make_probe_result() + probe.certificate = CertificateInfo(key_type="RSA", key_size=2048, curve=None) + monkeypatch.setattr( + "quantumvalidator.assessor.check_tls", lambda *a, **kw: probe + ) + report = assess("example.com") + cert_check = report.checks[2] + assert cert_check.name == "certificate_key" + assert cert_check.status == Status.FAIL + assert cert_check.value == "RSA-2048" + # PQC-02 failure does not flip the PQC-01 verdict. + assert report.verdict == Verdict.SAFE + + def test_ssh_host_keys_passed_to_checks(self, monkeypatch): + probe = make_probe_result( + tls_version="SSHv2", negotiated_group="mlkem768nistp256-sha256" + ) + probe.detected_starttls = "ssh" + probe.ssh_host_key_algorithms = ["ssh-ed25519", "ssh-rsa"] + monkeypatch.setattr( + "quantumvalidator.assessor.check_tls", lambda *a, **kw: probe + ) + report = assess("ssh.example.com", port=22) + hk_check = report.checks[2] + assert hk_check.name == "host_key_algorithms" + assert hk_check.status == Status.FAIL + assert hk_check.value == "ssh-rsa" + assert report.verdict == Verdict.SAFE def test_default_port_443(self, monkeypatch): captured: dict = {} diff --git a/tests/test_tls_utils.py b/tests/test_tls_utils.py index a61a8b5..40b9905 100644 --- a/tests/test_tls_utils.py +++ b/tests/test_tls_utils.py @@ -9,15 +9,19 @@ from quantumvalidator.constants import OPENSSL_BINARY, PROBE_GROUPS from quantumvalidator.tls_utils import ( + CertificateInfo, _build_cmd, _extract_connection_error, + _extract_pem_cert, _fingerprint_banner, + _parse_certificate, _parse_openssl_output, _parse_ssh_kexinit, _probe_ftp, _probe_ssh, _read_server_banner, _read_ssh_packet, + fetch_certificate, probe_raw, probe_tls, ) @@ -97,6 +101,61 @@ 'OK "Dovecot ready."\n' ) +# --------------------------------------------------------------------------- +# Static self-signed test certificates (synthetic, CN=test.example.com) +# --------------------------------------------------------------------------- + +_RSA2048_PEM = """-----BEGIN CERTIFICATE----- +MIICwjCCAaqgAwIBAgIUZDc6/Vj84hVqlq2Se1E8eYnT4NEwDQYJKoZIhvcNAQEL +BQAwGzEZMBcGA1UEAwwQdGVzdC5leGFtcGxlLmNvbTAeFw0yNjAxMDEwMDAwMDBa +Fw0zNTEyMzAwMDAwMDBaMBsxGTAXBgNVBAMMEHRlc3QuZXhhbXBsZS5jb20wggEi +MA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQDcBeWqaG1BWyH62nR57jewISI0 +8EcrWwNAJ/41qlR1WqXSPRtaOJIdaM/W7j/IgMqTaVdFffEuo1bMljspcN/RQ+VU +eIleURucAWywPWkEQhPY8yYRsRjCTNtLwvcytkz8bxTnmq5BxocBp0hf2wSp/QzC +Gg38YwfaeXsEN40c+F8WrpHisBKt9zdyLkIiqWXSdPbQjnfjWlfmwHush73NYAIX +6lslnAZIbE5G+O5v0YFWp+qK/fWTiwrqkpGkEY15AvU4lGoH8yL6FDD7ccK76cg4 +nKI/Vc77KcBW8hWt+zpY/K9BvIUrfArOHfggF/91LObDONP5PAgXM2y4BBXLAgMB +AAEwDQYJKoZIhvcNAQELBQADggEBADfgnBVWiSbZ9Cdh4umcLp2i7cYtZ9T3I7Nz +ygrLIcaDGBUa2W+wa9RFd1awnCkLj9PfAV6xVuNEvXGmpMPfFkJEIyKA2CzWmMUc +gKRZ/LW8HpE21W1Cjfz6y7y/gcOLezA7iOSCwmbTwwBxPTBSDqMyl1W1u9ZO3pF5 +K5VtTUutuN3QJHFOCWCqD+tzN+QIPG7AknnqfbTXq0LKWp6FhiAC3fZil3499RFE +4h8qjs2/A+K+eU0UBpfEf9Luup71PI/2y16a5aQoiknVxLxn4TAhiWajPO+iUDcH +pCybZdg+ukGfUvvgtU7XX0BnwzNlTDUT7d2mO8iC99sDNtsCk3g= +-----END CERTIFICATE-----""" + +_ECP256_PEM = """-----BEGIN CERTIFICATE----- +MIIBNjCB3KADAgECAhQUB1bQl2IrHZA+qezwTzuQKbGpUDAKBggqhkjOPQQDAjAb +MRkwFwYDVQQDDBB0ZXN0LmV4YW1wbGUuY29tMB4XDTI2MDEwMTAwMDAwMFoXDTM1 +MTIzMDAwMDAwMFowGzEZMBcGA1UEAwwQdGVzdC5leGFtcGxlLmNvbTBZMBMGByqG +SM49AgEGCCqGSM49AwEHA0IABFowMOl2V2yMLmgZPmbOl+5go3698asqSImU3gPF +e+8tEkGgWLUP7VSzWilyToS60vHVtHHwho8TYNXZXNYrMUwwCgYIKoZIzj0EAwID +SQAwRgIhALXpLiA3eWPb+ln5oIHCOThjo6onuSEIQOwOqmxCAbekAiEAneXkrCNi +9Z7xSYs8YgAYU5+HHg+Sa5pcS5LxZLKr2Ks= +-----END CERTIFICATE-----""" + +_ED25519_PEM = """-----BEGIN CERTIFICATE----- +MIH1MIGooAMCAQICFAaRh6r0i5nq0nwjsIVmmAxUwYZPMAUGAytlcDAbMRkwFwYD +VQQDDBB0ZXN0LmV4YW1wbGUuY29tMB4XDTI2MDEwMTAwMDAwMFoXDTM1MTIzMDAw +MDAwMFowGzEZMBcGA1UEAwwQdGVzdC5leGFtcGxlLmNvbTAqMAUGAytlcAMhAPU8 +3V4NtjFZZdRjb13deAZqqnafVU63Rwd0DgKbinZGMAUGAytlcANBADHCbY7Q9HIt +k7Bm7S/KfnNMeSVZlzAeB4sLYETcTuHsB39Pjtqw5zkpsiHArbQVHfHhf+Q77rJ8 +qFyqq8ahig0= +-----END CERTIFICATE-----""" + +# openssl s_client output without -brief: banner lines around a PEM block. +_RAW_CERT_OUTPUT = ( + "Connecting to 192.0.2.1\n" + "CONNECTED(00000003)\n" + "---\n" + "Certificate chain\n" + " 0 s:CN=test.example.com\n" + "---\n" + "Server certificate\n" + f"{_RSA2048_PEM}\n" + "subject=CN=test.example.com\n" + "---\n" +) + def _make_sock_ctx(data: bytes = b"") -> MagicMock: """Return a MagicMock context-manager that mimics socket.create_connection.""" @@ -107,27 +166,41 @@ def _make_sock_ctx(data: bytes = b"") -> MagicMock: return sock -def _make_kexinit_packet(kex_algorithms: list[str]) -> bytes: - """Build a minimal SSH_MSG_KEXINIT binary payload (RFC 4253 §7.1) for testing.""" +def _make_kexinit_packet( + kex_algorithms: list[str], + host_key_algorithms: list[str] | None = None, +) -> bytes: + """Build a minimal SSH_MSG_KEXINIT binary payload (RFC 4253 §7.1) for testing. + + When *host_key_algorithms* is None the packet is truncated after the + kex_algorithms name-list (the pre-0.7.0 fixture shape). + """ kex_str = ",".join(kex_algorithms).encode("ascii") kex_len = len(kex_str).to_bytes(4, "big") - return ( + packet = ( b"\x00" # padding_length + b"\x14" # msg_type = 20 = SSH_MSG_KEXINIT + b"\x00" * 16 # cookie (all zeros for tests) + kex_len + kex_str ) + if host_key_algorithms is not None: + hk_str = ",".join(host_key_algorithms).encode("ascii") + packet += len(hk_str).to_bytes(4, "big") + hk_str + return packet def _make_ssh_sock( kex_algs: list[str] | None = None, banner: bytes = b"SSH-2.0-OpenSSH_9.9\r\n", + host_key_algs: list[str] | None = None, ) -> MagicMock: """Return a MagicMock socket simulating an SSH-2.0 server for _probe_ssh.""" if kex_algs is None: kex_algs = ["mlkem768nistp256-sha256", "curve25519-sha256"] - packet = _make_kexinit_packet(kex_algs) + if host_key_algs is None: + host_key_algs = ["ssh-ed25519", "rsa-sha2-512"] + packet = _make_kexinit_packet(kex_algs, host_key_algs) length_bytes = len(packet).to_bytes(4, "big") data = banner + length_bytes + packet state = {"pos": 0} @@ -382,15 +455,53 @@ def test_host_and_port_in_result(self, monkeypatch): assert result.port == 8443 def test_input_is_empty_bytes(self, monkeypatch): - captured_kw: dict = {} + captured_inputs: list = [] def capture(*a, **kw): - captured_kw.update(kw) + captured_inputs.append(kw.get("input")) return _make_proc(stdout=_PQC_OUTPUT) monkeypatch.setattr("quantumvalidator.tls_utils.subprocess.run", capture) probe_tls("example.com", 443) - assert captured_kw.get("input") == b"" + # First call is the -brief probe (empty stdin); second is the + # certificate fetch via probe_raw (sends QUIT). + assert captured_inputs[0] == b"" + assert captured_inputs[1] == b"QUIT\r\n" + + def test_certificate_attached_on_success(self, monkeypatch): + outputs = [_PQC_OUTPUT, _RAW_CERT_OUTPUT] + + def fake_run(*a, **kw): + return _make_proc(stdout=outputs.pop(0)) + + monkeypatch.setattr("quantumvalidator.tls_utils.subprocess.run", fake_run) + result = probe_tls("example.com", 443) + assert result.ok + assert result.certificate == CertificateInfo( + key_type="RSA", key_size=2048, curve=None + ) + + def test_certificate_none_when_fetch_has_no_pem(self, monkeypatch): + monkeypatch.setattr( + "quantumvalidator.tls_utils.subprocess.run", + lambda *a, **kw: _make_proc(stdout=_PQC_OUTPUT), + ) + result = probe_tls("example.com", 443) + assert result.ok + assert result.certificate is None + + def test_no_certificate_fetch_on_probe_error(self, monkeypatch): + calls: list = [] + + def fake_run(cmd, **kw): + calls.append(list(cmd)) + return _make_proc(stderr=_REFUSED_OUTPUT) + + monkeypatch.setattr("quantumvalidator.tls_utils.subprocess.run", fake_run) + result = probe_tls("example.com", 443) + assert not result.ok + assert len(calls) == 1 + assert result.certificate is None # --------------------------------------------------------------------------- @@ -602,7 +713,8 @@ def test_sieve_banner_embedded_in_output(self): class TestBannerFirstProbe: - """Banner-first probe: one socket read, one openssl call — always two connections.""" + """Banner-first probe: one socket read, one -brief openssl call, and on + success one more openssl call (no -brief) for the certificate fetch.""" def test_no_banner_uses_raw_tls(self, monkeypatch): calls: list = [] @@ -618,8 +730,10 @@ def fake_run(cmd, **kw): ) result = probe_tls("cloudflare.com", 443) assert result.ok - assert len(calls) == 1 + assert len(calls) == 2 assert "-starttls" not in calls[0] + assert "-brief" in calls[0] + assert "-brief" not in calls[1] # certificate fetch needs cert output assert result.detected_starttls is None def test_smtp_banner_single_starttls_probe(self, monkeypatch): @@ -635,9 +749,10 @@ def fake_run(cmd, **kw): lambda *a, **kw: _make_sock_ctx(b"220 smtp.gmail.com ESMTP\r\n"), ) result = probe_tls("smtp.gmail.com", 587) - assert len(calls) == 1 + assert len(calls) == 2 assert "-starttls" in calls[0] assert "smtp" in calls[0] + assert "-starttls" in calls[1] # certificate fetch reuses the mode assert result.detected_starttls == "smtp" assert result.ok @@ -654,7 +769,7 @@ def fake_run(cmd, **kw): lambda *a, **kw: _make_sock_ctx(b"* OK Dovecot ready.\r\n"), ) result = probe_tls("mail.example.com", 143) - assert len(calls) == 1 + assert len(calls) == 2 assert result.detected_starttls == "imap" assert result.ok @@ -671,7 +786,7 @@ def fake_run(cmd, **kw): lambda *a, **kw: _make_sock_ctx(b"+OK Dovecot ready.\r\n"), ) result = probe_tls("mail.example.com", 110) - assert len(calls) == 1 + assert len(calls) == 2 assert result.detected_starttls == "pop3" assert result.ok @@ -743,7 +858,7 @@ def fake_socket(*a, **kw): monkeypatch.setattr("quantumvalidator.tls_utils.subprocess.run", fake_run) monkeypatch.setattr("quantumvalidator.tls_utils.socket.create_connection", fake_socket) result = probe_tls("ftp.example.com", 21) - assert len(calls) == 1 + assert len(calls) == 2 assert "-starttls" in calls[0] assert "ftp" in calls[0] assert result.detected_starttls == "ftp" @@ -784,7 +899,7 @@ def fake_run(cmd, **kw): lambda *a, **kw: _make_sock_ctx(b"220 mail.example.com LMTP Postfix\r\n"), ) result = probe_tls("lmtp.example.com", 24) - assert len(calls) == 1 + assert len(calls) == 2 assert "-starttls" in calls[0] assert "lmtp" in calls[0] assert result.detected_starttls == "lmtp" @@ -803,7 +918,7 @@ def fake_run(cmd, **kw): lambda *a, **kw: _make_sock_ctx(b"200 news.example.com InterNetNews ready\r\n"), ) result = probe_tls("news.example.com", 119) - assert len(calls) == 1 + assert len(calls) == 2 assert "-starttls" in calls[0] assert "nntp" in calls[0] assert result.detected_starttls == "nntp" @@ -822,7 +937,7 @@ def fake_run(cmd, **kw): lambda *a, **kw: _make_sock_ctx(b'"IMPLEMENTATION" "Dovecot Pigeonhole"\r\n'), ) result = probe_tls("sieve.example.com", 4190) - assert len(calls) == 1 + assert len(calls) == 2 assert "-starttls" in calls[0] assert "sieve" in calls[0] assert result.detected_starttls == "sieve" @@ -982,20 +1097,22 @@ def test_fragmented_header_reassembled(self): class TestParseSshKexinit: def test_single_pqc_algorithm(self): packet = _make_kexinit_packet(["mlkem768nistp256-sha256"]) - assert _parse_ssh_kexinit(packet) == ["mlkem768nistp256-sha256"] + kex, _ = _parse_ssh_kexinit(packet) + assert kex == ["mlkem768nistp256-sha256"] def test_multiple_algorithms_parsed(self): algs = ["mlkem768nistp256-sha256", "curve25519-sha256", "ecdh-sha2-nistp256"] packet = _make_kexinit_packet(algs) - assert _parse_ssh_kexinit(packet) == algs + kex, _ = _parse_ssh_kexinit(packet) + assert kex == algs def test_wrong_message_type_returns_empty(self): packet = _make_kexinit_packet(["mlkem768nistp256-sha256"]) bad = packet[0:1] + b"\x15" + packet[2:] - assert _parse_ssh_kexinit(bad) == [] + assert _parse_ssh_kexinit(bad) == ([], []) def test_too_short_packet_returns_empty(self): - assert _parse_ssh_kexinit(b"\x00\x14" + b"\x00" * 10) == [] + assert _parse_ssh_kexinit(b"\x00\x14" + b"\x00" * 10) == ([], []) def test_truncated_name_list_returns_empty(self): packet = ( @@ -1004,7 +1121,25 @@ def test_truncated_name_list_returns_empty(self): + (100).to_bytes(4, "big") + b"hello" ) - assert _parse_ssh_kexinit(packet) == [] + assert _parse_ssh_kexinit(packet) == ([], []) + + def test_host_key_algorithms_parsed(self): + packet = _make_kexinit_packet( + ["curve25519-sha256"], ["ssh-ed25519", "rsa-sha2-512", "ssh-rsa"] + ) + kex, host_keys = _parse_ssh_kexinit(packet) + assert kex == ["curve25519-sha256"] + assert host_keys == ["ssh-ed25519", "rsa-sha2-512", "ssh-rsa"] + + def test_missing_host_key_list_returns_empty_host_keys(self): + packet = _make_kexinit_packet(["curve25519-sha256"]) + assert _parse_ssh_kexinit(packet) == (["curve25519-sha256"], []) + + def test_truncated_host_key_list_returns_empty_host_keys(self): + packet = _make_kexinit_packet(["curve25519-sha256"]) + ( + (100).to_bytes(4, "big") + b"ssh-ed" + ) + assert _parse_ssh_kexinit(packet) == (["curve25519-sha256"], []) # --------------------------------------------------------------------------- @@ -1024,6 +1159,27 @@ def test_pqc_kex_detected(self, monkeypatch): assert result.negotiated_group == "mlkem768nistp256-sha256" assert result.error is None assert result.detected_starttls == "ssh" + assert result.ssh_host_key_algorithms == ["ssh-ed25519", "rsa-sha2-512"] + + def test_host_key_algorithms_captured(self, monkeypatch): + monkeypatch.setattr( + "quantumvalidator.tls_utils.socket.create_connection", + lambda *a, **kw: _make_ssh_sock( + host_key_algs=["ssh-ed25519", "ssh-rsa", "ssh-dss"] + ), + ) + result = _probe_ssh("ssh.example.com", 22, 10) + assert result.ssh_host_key_algorithms == ["ssh-ed25519", "ssh-rsa", "ssh-dss"] + + def test_missing_host_key_list_gives_none(self, monkeypatch): + # Packet truncated after the kex name-list (pre-RFC-complete server). + monkeypatch.setattr( + "quantumvalidator.tls_utils.socket.create_connection", + lambda *a, **kw: _make_ssh_sock(host_key_algs=[]), + ) + result = _probe_ssh("ssh.example.com", 22, 10) + assert result.ok + assert result.ssh_host_key_algorithms is None def test_classical_only_returns_first_alg(self, monkeypatch): algs = ["curve25519-sha256", "ecdh-sha2-nistp256"] @@ -1277,3 +1433,114 @@ def test_non_utf8_output_decoded_with_replace(self, monkeypatch): assert result is not None assert "ok" in result + + +# --------------------------------------------------------------------------- +# _extract_pem_cert / _parse_certificate / fetch_certificate (PQC-02) +# --------------------------------------------------------------------------- + + +class TestExtractPemCert: + def test_pem_block_extracted(self): + pem = _extract_pem_cert(_RAW_CERT_OUTPUT) + assert pem is not None + assert pem.startswith("-----BEGIN CERTIFICATE-----") + assert pem.endswith("-----END CERTIFICATE-----") + + def test_no_pem_returns_none(self): + assert _extract_pem_cert(_PQC_OUTPUT) is None + + def test_first_block_taken_when_multiple(self): + output = _RSA2048_PEM + "\n" + _ECP256_PEM + "\n" + assert _extract_pem_cert(output) == _RSA2048_PEM + + +class TestParseCertificate: + def test_rsa_2048(self): + info = _parse_certificate(_RSA2048_PEM) + assert info == CertificateInfo(key_type="RSA", key_size=2048, curve=None) + + def test_ec_p256(self): + info = _parse_certificate(_ECP256_PEM) + assert info == CertificateInfo(key_type="EC", key_size=256, curve="secp256r1") + + def test_ed25519(self): + info = _parse_certificate(_ED25519_PEM) + assert info == CertificateInfo(key_type="Ed25519", key_size=256, curve=None) + + def test_invalid_pem_returns_none(self): + bad = "-----BEGIN CERTIFICATE-----\nnot base64!!\n-----END CERTIFICATE-----" + assert _parse_certificate(bad) is None + + def test_dsa_key_recognised(self, monkeypatch): + from cryptography.hazmat.primitives.asymmetric import dsa + + fake_key = MagicMock(spec=dsa.DSAPublicKey) + fake_key.key_size = 1024 + fake_cert = MagicMock() + fake_cert.public_key.return_value = fake_key + monkeypatch.setattr( + "quantumvalidator.tls_utils.x509.load_pem_x509_certificate", + lambda *a, **kw: fake_cert, + ) + info = _parse_certificate(_RSA2048_PEM) + assert info == CertificateInfo(key_type="DSA", key_size=1024, curve=None) + + def test_ed448_key_recognised(self, monkeypatch): + from cryptography.hazmat.primitives.asymmetric import ed448 + + fake_key = MagicMock(spec=ed448.Ed448PublicKey) + fake_cert = MagicMock() + fake_cert.public_key.return_value = fake_key + monkeypatch.setattr( + "quantumvalidator.tls_utils.x509.load_pem_x509_certificate", + lambda *a, **kw: fake_cert, + ) + info = _parse_certificate(_ED25519_PEM) + assert info == CertificateInfo(key_type="Ed448", key_size=456, curve=None) + + def test_unknown_key_type(self, monkeypatch): + fake_cert = MagicMock() + fake_cert.public_key.return_value = object() + monkeypatch.setattr( + "quantumvalidator.tls_utils.x509.load_pem_x509_certificate", + lambda *a, **kw: fake_cert, + ) + info = _parse_certificate(_RSA2048_PEM) + assert info == CertificateInfo(key_type="unknown", key_size=None, curve=None) + + +class TestFetchCertificate: + def test_returns_key_details(self, monkeypatch): + monkeypatch.setattr( + "quantumvalidator.tls_utils.subprocess.run", + lambda *a, **kw: _make_raw_proc(stdout=_RAW_CERT_OUTPUT.encode()), + ) + info = fetch_certificate("example.com", 443) + assert info == CertificateInfo(key_type="RSA", key_size=2048, curve=None) + + def test_probe_failure_returns_none(self, monkeypatch): + def raise_oserror(*a, **kw): + raise OSError("Network unreachable") + + monkeypatch.setattr("quantumvalidator.tls_utils.subprocess.run", raise_oserror) + assert fetch_certificate("example.com", 443) is None + + def test_output_without_pem_returns_none(self, monkeypatch): + monkeypatch.setattr( + "quantumvalidator.tls_utils.subprocess.run", + lambda *a, **kw: _make_raw_proc(stdout=_PQC_OUTPUT.encode()), + ) + assert fetch_certificate("example.com", 443) is None + + def test_starttls_mode_passed_through(self, monkeypatch): + captured: dict = {} + + def capture(cmd, **kw): + captured["cmd"] = list(cmd) + return _make_raw_proc(stdout=_RAW_CERT_OUTPUT.encode()) + + monkeypatch.setattr("quantumvalidator.tls_utils.subprocess.run", capture) + fetch_certificate("mail.example.com", 587, starttls="smtp") + assert "-starttls" in captured["cmd"] + assert "smtp" in captured["cmd"] diff --git a/tests/test_verdict.py b/tests/test_verdict.py index c7169eb..a27cc72 100644 --- a/tests/test_verdict.py +++ b/tests/test_verdict.py @@ -3,6 +3,7 @@ from __future__ import annotations from quantumvalidator.models import Status, Verdict +from quantumvalidator.tls_utils import CertificateInfo from quantumvalidator.verdict import build_checks, determine_verdict @@ -34,13 +35,15 @@ def test_unsafe_tls13_p256(self): class TestBuildChecks: - def test_safe_produces_two_checks(self): + def test_safe_produces_three_checks(self): checks = build_checks("TLSv1.3", "X25519MLKEM768") - assert len(checks) == 2 + assert len(checks) == 3 - def test_safe_both_pass(self): + def test_safe_first_two_pass_cert_info_when_missing(self): checks = build_checks("TLSv1.3", "X25519MLKEM768") - assert all(c.status == Status.PASS for c in checks) + assert all(c.status == Status.PASS for c in checks[:2]) + assert checks[2].name == "certificate_key" + assert checks[2].status == Status.INFO def test_tls12_first_check_fail(self): checks = build_checks("TLSv1.2", None) @@ -92,13 +95,15 @@ def test_unsafe_no_group(self): class TestBuildChecksSsh: - def test_pqc_kex_produces_two_checks(self): + def test_pqc_kex_produces_three_checks(self): checks = build_checks("SSHv2", "mlkem768nistp256-sha256") - assert len(checks) == 2 + assert len(checks) == 3 - def test_pqc_kex_both_pass(self): + def test_pqc_kex_first_two_pass_host_keys_info_when_missing(self): checks = build_checks("SSHv2", "mlkem768nistp256-sha256") - assert all(c.status == Status.PASS for c in checks) + assert all(c.status == Status.PASS for c in checks[:2]) + assert checks[2].name == "host_key_algorithms" + assert checks[2].status == Status.INFO def test_ssh_version_check_first(self): checks = build_checks("SSHv2", "mlkem768nistp256-sha256") @@ -116,3 +121,164 @@ def test_classical_kex_second_check_fail(self): def test_ssh_version_check_value(self): checks = build_checks("SSHv2", "mlkem768nistp256-sha256") assert checks[0].value == "SSHv2" + + +class TestCertificateCheck: + """PQC-02 — certificate_key check built by build_checks for TLS probes.""" + + @staticmethod + def _cert_check(certificate): + checks = build_checks("TLSv1.3", "X25519MLKEM768", certificate=certificate) + assert checks[2].name == "certificate_key" + return checks[2] + + def test_missing_certificate_is_info(self): + check = self._cert_check(None) + assert check.status == Status.INFO + assert check.value is None + + def test_rsa_3072_passes(self): + check = self._cert_check(CertificateInfo("RSA", 3072, None)) + assert check.status == Status.PASS + assert check.value == "RSA-3072" + assert "CNSA 2.0" in check.standard + + def test_rsa_4096_passes(self): + check = self._cert_check(CertificateInfo("RSA", 4096, None)) + assert check.status == Status.PASS + assert check.value == "RSA-4096" + + def test_rsa_2048_fails(self): + check = self._cert_check(CertificateInfo("RSA", 2048, None)) + assert check.status == Status.FAIL + assert check.value == "RSA-2048" + assert "3072" in check.reason + + def test_rsa_1024_fails(self): + check = self._cert_check(CertificateInfo("RSA", 1024, None)) + assert check.status == Status.FAIL + + def test_rsa_unknown_size_fails(self): + check = self._cert_check(CertificateInfo("RSA", None, None)) + assert check.status == Status.FAIL + + def test_ec_p256_passes(self): + check = self._cert_check(CertificateInfo("EC", 256, "secp256r1")) + assert check.status == Status.PASS + assert check.value == "P-256" + assert "SP 800-186" in check.standard + + def test_ec_p384_passes(self): + check = self._cert_check(CertificateInfo("EC", 384, "secp384r1")) + assert check.status == Status.PASS + assert check.value == "P-384" + + def test_ec_brainpool256_passes(self): + check = self._cert_check(CertificateInfo("EC", 256, "brainpoolP256r1")) + assert check.status == Status.PASS + assert check.value == "brainpoolP256r1" + + def test_ec_p224_fails(self): + check = self._cert_check(CertificateInfo("EC", 224, "secp224r1")) + assert check.status == Status.FAIL + assert check.value == "P-224" + + def test_ec_secp256k1_fails(self): + check = self._cert_check(CertificateInfo("EC", 256, "secp256k1")) + assert check.status == Status.FAIL + assert check.value == "secp256k1" + + def test_ec_unknown_curve_fails(self): + check = self._cert_check(CertificateInfo("EC", None, None)) + assert check.status == Status.FAIL + assert check.value == "unknown" + + def test_ed25519_passes(self): + check = self._cert_check(CertificateInfo("Ed25519", 256, None)) + assert check.status == Status.PASS + assert check.value == "Ed25519" + + def test_ed448_passes(self): + check = self._cert_check(CertificateInfo("Ed448", 456, None)) + assert check.status == Status.PASS + + def test_dsa_fails(self): + check = self._cert_check(CertificateInfo("DSA", 1024, None)) + assert check.status == Status.FAIL + assert check.value == "DSA-1024" + assert "FIPS 186-5" in check.standard + + def test_unknown_key_type_is_info(self): + check = self._cert_check(CertificateInfo("unknown", None, None)) + assert check.status == Status.INFO + assert check.value == "unknown" + + def test_weak_cert_does_not_change_verdict(self): + # PQC-02 failures carry their own weight; verdict tracks PQC-01 only. + assert determine_verdict("TLSv1.3", "X25519MLKEM768") == Verdict.SAFE + checks = build_checks( + "TLSv1.3", "X25519MLKEM768", + certificate=CertificateInfo("RSA", 2048, None), + ) + assert checks[2].status == Status.FAIL + + +class TestHostKeyCheck: + """PQC-03 — host_key_algorithms check built by build_checks for SSH probes.""" + + @staticmethod + def _hk_check(algs): + checks = build_checks( + "SSHv2", "mlkem768nistp256-sha256", ssh_host_key_algorithms=algs + ) + assert checks[2].name == "host_key_algorithms" + return checks[2] + + def test_none_is_info(self): + check = self._hk_check(None) + assert check.status == Status.INFO + assert check.value is None + + def test_empty_list_is_info(self): + check = self._hk_check([]) + assert check.status == Status.INFO + + def test_clean_list_passes(self): + algs = ["ssh-ed25519", "rsa-sha2-512", "rsa-sha2-256", "ecdsa-sha2-nistp256"] + check = self._hk_check(algs) + assert check.status == Status.PASS + assert check.value == ",".join(algs) + assert "RFC 8332" in check.standard + + def test_ssh_rsa_fails(self): + check = self._hk_check(["ssh-ed25519", "ssh-rsa"]) + assert check.status == Status.FAIL + assert check.value == "ssh-rsa" + assert "SHA-1" in check.reason + + def test_ssh_dss_fails(self): + check = self._hk_check(["ssh-dss"]) + assert check.status == Status.FAIL + assert check.value == "ssh-dss" + assert "FIPS 186-5" in check.reason + + def test_multiple_deprecated_all_listed(self): + check = self._hk_check(["ssh-rsa", "ssh-dss", "ssh-ed25519"]) + assert check.status == Status.FAIL + assert check.value == "ssh-rsa,ssh-dss" + + def test_cert_v01_variants_fail(self): + check = self._hk_check(["ssh-rsa-cert-v01@openssh.com"]) + assert check.status == Status.FAIL + + def test_rsa_sha2_variants_do_not_fail(self): + check = self._hk_check(["rsa-sha2-256", "rsa-sha2-512"]) + assert check.status == Status.PASS + + def test_deprecated_host_key_does_not_change_verdict(self): + assert determine_verdict("SSHv2", "mlkem768nistp256-sha256") == Verdict.SAFE + checks = build_checks( + "SSHv2", "mlkem768nistp256-sha256", + ssh_host_key_algorithms=["ssh-rsa"], + ) + assert checks[2].status == Status.FAIL