diff --git a/docs/ENTERPRISE_SECURITY.md b/docs/ENTERPRISE_SECURITY.md index 84ed025c..5d7a784a 100644 --- a/docs/ENTERPRISE_SECURITY.md +++ b/docs/ENTERPRISE_SECURITY.md @@ -124,6 +124,19 @@ for Sealed Secrets / External Secrets input). Prefer `kubectl create secret For zero-downtime rotation, publish the new public key to verifiers first, then cut the manager over to the new private key. +### Deployment-domain tokens + +Long-lived **deployment-domain** tokens (used by the client-config pull flow) are +signed with the *same* asymmetric scheme — the manager's private key (ES256/RS256), +standard `iat`/`exp` claims, and `iss`/`aud`. They are verified with the public key +plus a database cross-check; HS256/none are rejected. + +> **Upgrade note:** deployment-domain tokens minted before `v2.1.x` were HS256. +> After upgrading, roll each domain's token over (Domain-Admin `rollover_jwt` +> permission / `rollover_domain_jwt`) to re-issue it under the asymmetric key. +> This also fixes a latent multi-replica bug where a per-process random secret +> made tokens unverifiable across manager replicas/restarts. + --- ## 2. Tenant isolation diff --git a/manager/backend/app/__init__.py b/manager/backend/app/__init__.py index e6e36acf..b2bd66a0 100644 --- a/manager/backend/app/__init__.py +++ b/manager/backend/app/__init__.py @@ -68,9 +68,18 @@ def create_app(config_class: type = Config) -> Flask: from app.services.whois_service import WHOISManager app.whois_manager = WHOISManager(db_url=app.config['DB_URL']) - # Initialize Client Config manager + # Initialize Client Config manager. Deployment-domain tokens are signed + # asymmetrically with the manager's configured JWT keypair (ES256 default, + # RS256 fallback) so they verify across replicas and restarts. from app.services.client_config_service import ClientConfigManager - app.client_config_manager = ClientConfigManager(db_url=app.config['DB_URL']) + app.client_config_manager = ClientConfigManager( + db_url=app.config['DB_URL'], + private_key=app.config.get('JWT_PRIVATE_KEY'), + public_key=app.config.get('JWT_PUBLIC_KEY'), + algorithm=app.config.get('JWT_ALGORITHM', 'ES256'), + issuer=app.config.get('JWT_ISSUER', 'squawk-manager'), + audience=app.config.get('JWT_AUDIENCE', 'squawk'), + ) # Register blueprints from app.blueprints.auth import auth_bp diff --git a/manager/backend/app/services/client_config_service.py b/manager/backend/app/services/client_config_service.py index 1ad2c4b3..ac3942d5 100644 --- a/manager/backend/app/services/client_config_service.py +++ b/manager/backend/app/services/client_config_service.py @@ -14,9 +14,8 @@ import json import logging -import secrets from dataclasses import dataclass, field -from datetime import datetime, timedelta +from datetime import datetime, timedelta, timezone from typing import Any, Dict, List, Optional import jwt @@ -24,6 +23,35 @@ logger = logging.getLogger(__name__) +# Verifiers accept asymmetric algorithms only. HS256/none are rejected to block +# the public-key-as-HMAC algorithm-confusion attack, consistent with the rest of +# the platform (dns-server/dhcp-server/ntp-server verifiers). +_DOMAIN_JWT_ALGORITHMS = ["ES256", "RS256"] + + +def _generate_ephemeral_es256_keypair() -> tuple[str, str]: + """Generate an in-process ES256 keypair (PEM strings). + + Used only when no manager keypair is supplied (standalone/dev/test). In a + real deployment the manager passes its configured JWT_PRIVATE_KEY / + JWT_PUBLIC_KEY so domain tokens are verifiable across replicas and restarts. + """ + from cryptography.hazmat.primitives.asymmetric import ec + from cryptography.hazmat.primitives import serialization + from cryptography.hazmat.backends import default_backend + + private_key = ec.generate_private_key(ec.SECP256R1(), default_backend()) + private_pem = private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ).decode("utf-8") + public_pem = private_key.public_key().public_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PublicFormat.SubjectPublicKeyInfo, + ).decode("utf-8") + return private_pem, public_pem + @dataclass(slots=True) class ConfigData: @@ -57,15 +85,44 @@ class OverrideRecord: class ClientConfigManager: """Manages client configurations using penguin-dal.""" - def __init__(self, db_url: str, jwt_secret: Optional[str] = None) -> None: + def __init__( + self, + db_url: str, + private_key: Optional[str] = None, + public_key: Optional[str] = None, + algorithm: str = "ES256", + issuer: str = "squawk-manager", + audience: str = "squawk", + ) -> None: """Initialize client config manager. + Deployment-domain tokens are signed asymmetrically (ES256 default, + RS256 fallback) with the manager's private key and verified with the + public key — matching the platform-wide JWT scheme. Supply the manager's + configured keypair so tokens remain valid across replicas/restarts; if + omitted, an ephemeral keypair is generated for the process (dev/test). + Args: db_url: Database connection string - jwt_secret: JWT signing secret (generated if not provided) + private_key: PEM private key used to sign domain tokens + public_key: PEM public key used to verify domain tokens + algorithm: Signing algorithm (ES256 or RS256) + issuer: Expected/emitted `iss` claim + audience: Expected/emitted `aud` claim """ self.db_url = db_url - self.jwt_secret = jwt_secret or secrets.token_urlsafe(32) + if not private_key or not public_key: + private_key, public_key = _generate_ephemeral_es256_keypair() + logger.warning( + "ClientConfigManager: no JWT keypair supplied; using an " + "ephemeral in-process keypair. Domain tokens will not be " + "verifiable across replicas or restarts." + ) + self.private_key = private_key + self.public_key = public_key + self.algorithm = algorithm if algorithm in _DOMAIN_JWT_ALGORITHMS else "ES256" + self.issuer = issuer + self.audience = audience self._initialize_default_roles() def _get_db(self) -> DB: @@ -73,14 +130,21 @@ def _get_db(self) -> DB: return DB(self.db_url) def _generate_domain_jwt(self, domain_name: str) -> str: - """Generate JWT token for deployment domain.""" + """Generate an asymmetrically-signed JWT for a deployment domain. + + Signed with the manager's private key (ES256/RS256). Uses standard + `iat`/`exp` claims so PyJWT enforces expiry natively, plus `iss`/`aud`. + """ + now = datetime.now(timezone.utc) payload = { "domain": domain_name, "type": "deployment_domain", - "issued_at": datetime.now().timestamp(), - "expires_at": (datetime.now() + timedelta(days=365)).timestamp(), + "iss": self.issuer, + "aud": self.audience, + "iat": now, + "exp": now + timedelta(days=365), } - return jwt.encode(payload, self.jwt_secret, algorithm="HS256") + return jwt.encode(payload, self.private_key, algorithm=self.algorithm) def _validate_config_data(self, config_data: Dict[str, Any]) -> bool: """Validate client configuration data structure. @@ -329,13 +393,16 @@ def _verify_domain_jwt(self, jwt_token: str) -> Optional[Dict[str, Any]]: """Verify domain JWT token and return domain info.""" db = self._get_db() try: - # Decode JWT - payload = jwt.decode(jwt_token, self.jwt_secret, algorithms=["HS256"]) - - # Check if token is expired - expires_at = payload.get("expires_at") - if expires_at and datetime.now().timestamp() > expires_at: - return None + # Verify signature with the public key. Expiry (`exp`) is enforced + # natively by PyJWT; iss/aud are validated. HS256/none are rejected. + payload = jwt.decode( + jwt_token, + self.public_key, + algorithms=_DOMAIN_JWT_ALGORITHMS, + issuer=self.issuer, + audience=self.audience, + options={"require": ["exp", "iat"]}, + ) domain_name = payload.get("domain") if not domain_name: diff --git a/manager/backend/tests/test_client_config_domain_jwt.py b/manager/backend/tests/test_client_config_domain_jwt.py new file mode 100644 index 00000000..e11b2d5b --- /dev/null +++ b/manager/backend/tests/test_client_config_domain_jwt.py @@ -0,0 +1,215 @@ +"""Regression tests: deployment-domain JWTs are asymmetrically signed. + +Domain tokens minted by ClientConfigManager must be signed with the manager's +private key (ES256 default / RS256 fallback) and verified with the public key. +HS256/none and wrong-key tokens must be rejected (algorithm-confusion defense), +matching the platform-wide JWT scheme. +""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + +import jwt +import pytest +from cryptography.hazmat.backends import default_backend +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import ec + +from app.services.client_config_service import ClientConfigManager + + +def _gen_es256_pem() -> tuple[str, str]: + key = ec.generate_private_key(ec.SECP256R1(), default_backend()) + priv = key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ).decode() + pub = key.public_key().public_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PublicFormat.SubjectPublicKeyInfo, + ).decode() + return priv, pub + + +class _FakeField: + """Supports the ``==`` / ``&`` operators used to build penguin-dal queries.""" + + def __eq__(self, other): # noqa: D105 + return self + + def __and__(self, other): # noqa: D105 + return self + + __rand__ = __and__ + + +class _FakeSelect: + def __init__(self, row): + self._row = row + + def first(self): + return self._row + + +class _FakeQuery: + def __init__(self, row): + self._row = row + + def select(self): + return _FakeSelect(self._row) + + +class _FakeTable: + def __getattr__(self, _name): + return _FakeField() + + +class _FakeDB: + """Minimal penguin-dal stand-in: returns a preset domain row from queries.""" + + def __init__(self, row): + self._row = row + + def __getattr__(self, _name): + return _FakeTable() + + def __call__(self, _query): + return _FakeQuery(self._row) + + def close(self): + pass + + +class _Row: + def __init__(self, id_, name, description=""): + self.id = id_ + self.name = name + self.description = description + + +@pytest.fixture +def manager(monkeypatch): + """A ClientConfigManager with a real ES256 keypair and no DB side effects.""" + priv, pub = _gen_es256_pem() + monkeypatch.setattr( + ClientConfigManager, "_initialize_default_roles", lambda self: None + ) + mgr = ClientConfigManager( + db_url="sqlite://:memory:", private_key=priv, public_key=pub + ) + # Domain lookups return a matching active row for whatever token verifies. + mgr._get_db = lambda: _FakeDB(_Row(1, "prod-domain", "Production")) # type: ignore[attr-defined] + return mgr, priv, pub + + +def test_domain_jwt_signed_es256_not_hs256(manager): + mgr, _priv, pub = manager + token = mgr._generate_domain_jwt("prod-domain") + assert jwt.get_unverified_header(token)["alg"] == "ES256" + + payload = jwt.decode( + token, pub, algorithms=["ES256"], audience="squawk", issuer="squawk-manager" + ) + assert payload["domain"] == "prod-domain" + assert payload["type"] == "deployment_domain" + assert "exp" in payload and "iat" in payload + + +def test_valid_domain_jwt_verifies(manager): + mgr, _priv, _pub = manager + token = mgr._generate_domain_jwt("prod-domain") + result = mgr._verify_domain_jwt(token) + assert result is not None + assert result["name"] == "prod-domain" + + +def test_wrong_key_domain_jwt_rejected(manager): + """A token signed by a different private key must fail verification.""" + mgr, _priv, _pub = manager + attacker_priv, _attacker_pub = _gen_es256_pem() + now = datetime.now(timezone.utc) + forged = jwt.encode( + { + "domain": "prod-domain", + "type": "deployment_domain", + "iss": "squawk-manager", + "aud": "squawk", + "iat": now, + "exp": now + timedelta(days=1), + }, + attacker_priv, + algorithm="ES256", + ) + assert mgr._verify_domain_jwt(forged) is None + + +def test_hs256_domain_jwt_rejected(manager): + """HS256 tokens must be rejected — verifier allows ES256/RS256 only. + + This defeats the public-key-as-HMAC algorithm-confusion attack: even a + validly-formed HS256 token cannot pass the algorithm allowlist. + """ + mgr, _priv, _pub = manager + now = datetime.now(timezone.utc) + forged = jwt.encode( + { + "domain": "prod-domain", + "type": "deployment_domain", + "iss": "squawk-manager", + "aud": "squawk", + "iat": now, + "exp": now + timedelta(days=1), + }, + "attacker-hmac-secret", + algorithm="HS256", + ) + assert mgr._verify_domain_jwt(forged) is None + + +def test_expired_domain_jwt_rejected(manager): + mgr, priv, _pub = manager + now = datetime.now(timezone.utc) + expired = jwt.encode( + { + "domain": "prod-domain", + "type": "deployment_domain", + "iss": "squawk-manager", + "aud": "squawk", + "iat": now - timedelta(days=2), + "exp": now - timedelta(days=1), + }, + priv, + algorithm="ES256", + ) + assert mgr._verify_domain_jwt(expired) is None + + +def test_missing_exp_domain_jwt_rejected(manager): + """Tokens without a required exp claim must be rejected.""" + mgr, priv, _pub = manager + now = datetime.now(timezone.utc) + no_exp = jwt.encode( + { + "domain": "prod-domain", + "type": "deployment_domain", + "iss": "squawk-manager", + "aud": "squawk", + "iat": now, + }, + priv, + algorithm="ES256", + ) + assert mgr._verify_domain_jwt(no_exp) is None + + +def test_ephemeral_keypair_when_unconfigured(monkeypatch): + """With no keypair supplied, a working ephemeral ES256 pair is generated.""" + monkeypatch.setattr( + ClientConfigManager, "_initialize_default_roles", lambda self: None + ) + mgr = ClientConfigManager(db_url="sqlite://:memory:") + mgr._get_db = lambda: _FakeDB(_Row(1, "d", "d")) # type: ignore[attr-defined] + token = mgr._generate_domain_jwt("d") + assert mgr._verify_domain_jwt(token) is not None