diff --git a/dns-server/app/services/selective_router.py b/dns-server/app/services/selective_router.py index fd85cf7b..5e06a6c8 100644 --- a/dns-server/app/services/selective_router.py +++ b/dns-server/app/services/selective_router.py @@ -3,15 +3,9 @@ Implements per-user/group zone access control. """ import logging -import jwt as pyjwt -from jwt.exceptions import ( - InvalidSignatureError, ExpiredSignatureError, DecodeError, InvalidTokenError, - InvalidAudienceError, InvalidIssuerError, MissingRequiredClaimError -) from typing import Dict, List, Optional -from app.config import ( - JWT_PUBLIC_KEY, JWT_ISSUER, JWT_AUDIENCE -) +from app.config import JWT_PUBLIC_KEY +from app.utils.jwt_verify import verify_squawk_jwt logger = logging.getLogger(__name__) @@ -71,71 +65,47 @@ def check_zone_permission(self, domain: str, token: Optional[str] = None) -> boo logger.debug(f"Access denied to {visibility} zone {zone['name']}: no token provided") return False - # Verify JWT signature and extract payload - try: - # Fail closed: JWT_PUBLIC_KEY must be configured for non-public zones - if not JWT_PUBLIC_KEY: - logger.error("JWT_PUBLIC_KEY not configured; denying access to non-public zone") - return False - - # Verify with public key (ES256/RS256), require tenant claim - payload = pyjwt.decode( - token, - JWT_PUBLIC_KEY, - algorithms=['ES256', 'RS256'], - audience=JWT_AUDIENCE, - issuer=JWT_ISSUER, - options={'require': ['exp', 'iat', 'tenant']} + # Verify JWT via the shared verifier (ES256/RS256, iss/aud, required + # exp/iat/tenant, fail closed) — authorization stays here. + payload = verify_squawk_jwt(token, JWT_PUBLIC_KEY) + if payload is None: + logger.debug( + f"Access denied to {visibility} zone {zone['name']}: token verification failed" ) + return False + + user_teams = payload.get('team_roles', {}).keys() if payload.get('team_roles') else [] + + # Check if user's teams are in allowed teams + allowed_teams = zone.get('allowed_teams', []) - # Fail closed: tenant claim must be present and non-empty - if not payload.get('tenant'): - logger.debug(f"Access denied: token missing or empty tenant claim") + if visibility == 'internal': + # Internal: must be member of allowed teams + if not allowed_teams or any(team in allowed_teams for team in user_teams): + return True + else: + logger.debug(f"Access denied to internal zone {zone['name']}: user teams {user_teams} not in {allowed_teams}") return False - user_teams = payload.get('team_roles', {}).keys() if payload.get('team_roles') else [] - user_id = payload.get('user_id') - - # Check if user's teams are in allowed teams - allowed_teams = zone.get('allowed_teams', []) - - if visibility == 'internal': - # Internal: must be member of allowed teams - if not allowed_teams or any(team in allowed_teams for team in user_teams): - return True - else: - logger.debug(f"Access denied to internal zone {zone['name']}: user teams {user_teams} not in {allowed_teams}") - return False - - elif visibility == 'restricted': - # Restricted: must be in specific allowed teams - if any(team in allowed_teams for team in user_teams): - return True - else: - logger.debug(f"Access denied to restricted zone {zone['name']}: user teams {user_teams} not in {allowed_teams}") - return False - - elif visibility == 'private': - # Private: admin only (check for admin role in token) - is_admin = payload.get('role') == 'admin' - if is_admin: - return True - else: - logger.debug(f"Access denied to private zone {zone['name']}: user is not admin") - return False + elif visibility == 'restricted': + # Restricted: must be in specific allowed teams + if any(team in allowed_teams for team in user_teams): + return True + else: + logger.debug(f"Access denied to restricted zone {zone['name']}: user teams {user_teams} not in {allowed_teams}") + return False + elif visibility == 'private': + # Private: admin only (check for admin role in token) + is_admin = payload.get('role') == 'admin' + if is_admin: + return True else: - # Unknown visibility, deny + logger.debug(f"Access denied to private zone {zone['name']}: user is not admin") return False - except (InvalidSignatureError, ExpiredSignatureError, DecodeError, InvalidTokenError) as e: - logger.warning(f"Invalid JWT token for zone access: {e}") - return False - except (InvalidAudienceError, InvalidIssuerError, MissingRequiredClaimError) as e: - logger.warning(f"JWT claim validation failed: {e}") - return False - except Exception as e: - logger.error(f"Token parsing error: {e}") + else: + # Unknown visibility, deny return False def get_zone_records(self, domain: str) -> Optional[List[Dict]]: diff --git a/dns-server/app/utils/jwt_verify.py b/dns-server/app/utils/jwt_verify.py new file mode 100644 index 00000000..da2e5ff4 --- /dev/null +++ b/dns-server/app/utils/jwt_verify.py @@ -0,0 +1,81 @@ +"""Shared Squawk JWT verification (single implementation for dns-server). + +Centralizes the ES256/RS256 verification contract used by every dns-server +authorization path (selective_router, resilience): + +- asymmetric algorithms only (ES256/RS256) — HS256/none rejected, blocking + the public-key-as-HMAC algorithm-confusion attack +- issuer + audience validated; exp/iat/tenant required +- fail closed: no public key configured, or missing/empty tenant → reject + +Callers pass their configured public key and act on the returned payload; +authorization (zone visibility/team rules) stays with the caller. +""" + +from __future__ import annotations + +import logging +from typing import Optional + +import jwt as pyjwt +from jwt.exceptions import ( + DecodeError, + ExpiredSignatureError, + InvalidAudienceError, + InvalidIssuerError, + InvalidSignatureError, + InvalidTokenError, + MissingRequiredClaimError, +) + +from app.config import JWT_AUDIENCE, JWT_ISSUER + +logger = logging.getLogger(__name__) + + +def verify_squawk_jwt(token: str, public_key: Optional[str]) -> Optional[dict]: + """Verify a Squawk user JWT; return its payload or None (fail closed). + + Args: + token: The presented bearer token. + public_key: PEM public key to verify with (caller's configured key). + + Returns: + The verified payload dict, or None on any failure — unconfigured key, + bad signature/alg, expired, wrong iss/aud, missing/empty tenant. + """ + if not public_key: + logger.error("JWT_PUBLIC_KEY not configured; denying access") + return None + + if not token: + return None + + try: + payload = pyjwt.decode( + token, + public_key, + algorithms=['ES256', 'RS256'], + audience=JWT_AUDIENCE, + issuer=JWT_ISSUER, + options={'require': ['exp', 'iat', 'tenant']}, + ) + except (InvalidSignatureError, ExpiredSignatureError, DecodeError) as e: + logger.warning(f"Invalid JWT token: {e}") + return None + except (InvalidAudienceError, InvalidIssuerError, MissingRequiredClaimError) as e: + logger.warning(f"JWT claim validation failed: {e}") + return None + except InvalidTokenError as e: + logger.warning(f"Invalid token: {e}") + return None + except Exception as e: + logger.error(f"Token validation error: {e}") + return None + + # Fail closed: tenant claim must be present and non-empty + if not payload.get('tenant'): + logger.debug("Access denied: token missing or empty tenant claim") + return None + + return payload diff --git a/dns-server/app/utils/resilience.py b/dns-server/app/utils/resilience.py index c7456c20..be3962db 100644 --- a/dns-server/app/utils/resilience.py +++ b/dns-server/app/utils/resilience.py @@ -5,14 +5,10 @@ import logging from datetime import datetime, timedelta from typing import Optional -import jwt as pyjwt -from jwt.exceptions import ( - InvalidSignatureError, ExpiredSignatureError, DecodeError, InvalidTokenError, - InvalidAudienceError, InvalidIssuerError, MissingRequiredClaimError -) from app.services.manager_client import ManagerClient -from app.config import JWT_PUBLIC_KEY, JWT_ISSUER, JWT_AUDIENCE +from app.config import JWT_PUBLIC_KEY +from app.utils.jwt_verify import verify_squawk_jwt logger = logging.getLogger(__name__) @@ -117,47 +113,22 @@ def _check_zone_permission(self, zone: dict, token: Optional[str]) -> bool: if not token: return False - try: - # Fail closed: JWT_PUBLIC_KEY must be configured for non-public zones - if not JWT_PUBLIC_KEY: - logger.error("JWT_PUBLIC_KEY not configured; denying access to non-public zone") - return False - - # Verify JWT signature and extract payload (ES256/RS256), require tenant - payload = pyjwt.decode( - token, - JWT_PUBLIC_KEY, - algorithms=['ES256', 'RS256'], - audience=JWT_AUDIENCE, - issuer=JWT_ISSUER, - options={'require': ['exp', 'iat', 'tenant']} - ) - - # Fail closed: tenant claim must be present and non-empty - if not payload.get('tenant'): - logger.debug(f"Access denied: token missing or empty tenant claim") - return False - - user_teams = list(payload.get('team_roles', {}).keys()) if payload.get('team_roles') else [] - allowed_teams = zone.get('allowed_teams', []) + # Verify JWT via the shared verifier (ES256/RS256, iss/aud, required + # exp/iat/tenant, fail closed) — team authorization stays here. + payload = verify_squawk_jwt(token, JWT_PUBLIC_KEY) + if payload is None: + return False - # Check team membership - if not allowed_teams: - # No team restrictions - return True + user_teams = list(payload.get('team_roles', {}).keys()) if payload.get('team_roles') else [] + allowed_teams = zone.get('allowed_teams', []) - # User must be in at least one allowed team - return any(team in allowed_teams for team in user_teams) + # Check team membership + if not allowed_teams: + # No team restrictions + return True - except (InvalidSignatureError, ExpiredSignatureError, DecodeError, InvalidTokenError) as e: - logger.warning(f"Invalid JWT token for zone access: {e}") - return False - except (InvalidAudienceError, InvalidIssuerError, MissingRequiredClaimError) as e: - logger.warning(f"JWT claim validation failed: {e}") - return False - except Exception as e: - logger.error(f"Token validation error: {e}") - return False + # User must be in at least one allowed team + return any(team in allowed_teams for team in user_teams) def get_mode(self) -> str: """Get current operational mode.""" diff --git a/dns-server/tests_full_future/test_client_config_api.py b/dns-server/tests_full_future/test_client_config_api.py index a5617712..004c0fe1 100644 --- a/dns-server/tests_full_future/test_client_config_api.py +++ b/dns-server/tests_full_future/test_client_config_api.py @@ -14,10 +14,13 @@ class TestClientConfigManager: @pytest.fixture - def config_manager(self, temp_db, test_jwt_secret): - """Create client config manager instance with test database""" + def config_manager(self, temp_db): + """Create client config manager instance with test database. + + No keypair supplied -> ephemeral ES256 keypair (asymmetric signing). + """ db_url = f"sqlite://{temp_db._uri[9:]}" # Extract path from DAL URI - return ClientConfigManager(db_url, test_jwt_secret) + return ClientConfigManager(db_url) def test_create_deployment_domain(self, config_manager): """Test creating a new deployment domain""" @@ -34,7 +37,10 @@ def test_create_deployment_domain(self, config_manager): # Verify JWT token is valid token = result['jwt_token'] - decoded = jwt.decode(token, config_manager.jwt_secret, algorithms=['HS256']) + decoded = jwt.decode( + token, config_manager.public_key, algorithms=['ES256', 'RS256'], + audience=config_manager.audience, issuer=config_manager.issuer, + ) assert decoded['domain'] == "test-domain" assert decoded['type'] == 'deployment_domain' @@ -65,7 +71,10 @@ def test_rollover_domain_jwt(self, config_manager): # Verify new JWT is valid new_token = rollover_result['new_jwt'] - decoded = jwt.decode(new_token, config_manager.jwt_secret, algorithms=['HS256']) + decoded = jwt.decode( + new_token, config_manager.public_key, algorithms=['ES256', 'RS256'], + audience=config_manager.audience, issuer=config_manager.issuer, + ) assert decoded['domain'] == "rollover-test" def test_create_client_config(self, config_manager, mock_client_config): @@ -353,16 +362,21 @@ def test_cleanup_inactive_clients(self, config_manager): clients = config_manager.get_domain_clients(domain_result['id']) assert len(clients) == 0 - def test_expired_jwt_rejection(self, config_manager, test_jwt_secret): + def test_expired_jwt_rejection(self, config_manager): """Test that expired JWT tokens are rejected""" - # Create expired JWT manually + # Create a genuinely expired ES256 domain token (signed with the + # manager's own private key so only expiry causes the rejection) expired_payload = { 'domain': 'expired-test', 'type': 'deployment_domain', - 'issued_at': (datetime.now() - timedelta(days=2)).timestamp(), - 'expires_at': (datetime.now() - timedelta(days=1)).timestamp() # Expired + 'iss': config_manager.issuer, + 'aud': config_manager.audience, + 'iat': datetime.now() - timedelta(days=2), + 'exp': datetime.now() - timedelta(days=1), # Expired } - expired_jwt = jwt.encode(expired_payload, test_jwt_secret, algorithm='HS256') + expired_jwt = jwt.encode( + expired_payload, config_manager.private_key, algorithm='ES256' + ) # Try to register client with expired JWT register_result = config_manager.register_client( diff --git a/manager/backend/app/config.py b/manager/backend/app/config.py index ab1df8e7..065d9a63 100644 --- a/manager/backend/app/config.py +++ b/manager/backend/app/config.py @@ -99,24 +99,10 @@ def __init__(self) -> None: super().__init__() # If no keys configured, generate an ephemeral ES256 keypair in-memory if not self.JWT_PRIVATE_KEY or not self.JWT_PUBLIC_KEY: - 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() + from app.utils.crypto import generate_ephemeral_es256_keypair + self.JWT_PRIVATE_KEY, self.JWT_PUBLIC_KEY = ( + generate_ephemeral_es256_keypair() ) - self.JWT_PRIVATE_KEY = private_key.private_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PrivateFormat.PKCS8, - encryption_algorithm=serialization.NoEncryption() - ).decode('utf-8') - - public_key = private_key.public_key() - self.JWT_PUBLIC_KEY = public_key.public_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PublicFormat.SubjectPublicKeyInfo - ).decode('utf-8') class ProductionConfig(Config): @@ -156,24 +142,10 @@ def __init__(self) -> None: """Generate ephemeral ES256 keypair for testing.""" super().__init__() # Always generate a fresh ES256 keypair for tests - 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() + from app.utils.crypto import generate_ephemeral_es256_keypair + self.JWT_PRIVATE_KEY, self.JWT_PUBLIC_KEY = ( + generate_ephemeral_es256_keypair() ) - self.JWT_PRIVATE_KEY = private_key.private_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PrivateFormat.PKCS8, - encryption_algorithm=serialization.NoEncryption() - ).decode('utf-8') - - public_key = private_key.public_key() - self.JWT_PUBLIC_KEY = public_key.public_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PublicFormat.SubjectPublicKeyInfo - ).decode('utf-8') # Config dictionary diff --git a/manager/backend/app/services/client_config_service.py b/manager/backend/app/services/client_config_service.py index ac3942d5..7548e9dc 100644 --- a/manager/backend/app/services/client_config_service.py +++ b/manager/backend/app/services/client_config_service.py @@ -12,9 +12,9 @@ from __future__ import annotations -import json import logging -from dataclasses import dataclass, field +from dataclasses import dataclass +from functools import wraps from datetime import datetime, timedelta, timezone from typing import Any, Dict, List, Optional @@ -29,30 +29,6 @@ _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: """Validated client configuration data.""" @@ -82,6 +58,49 @@ class OverrideRecord: expires_at: Optional[datetime] = None + +def _generate_ephemeral_es256_keypair() -> tuple[str, str]: + """Generate an in-process ES256 keypair (PEM strings). + + Intentionally self-contained (duplicates app.utils.crypto): this module is + imported BARE (as ``client_config_service``) by the acceptance-test harness + with the services dir on sys.path, where ``app.*`` resolves to a different + package — so it must not import from ``app.*``. + """ + 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 + + +def _with_db(method): + """Open a DB connection for the wrapped method and guarantee it closes. + + The wrapped method receives the connection as its first argument after + ``self``. Centralizes the get/close lifecycle previously duplicated in + every method (``db = self._get_db()`` ... ``finally: db.close()``). + """ + @wraps(method) + def wrapper(self, *args, **kwargs): + db = self._get_db() + try: + return method(self, db, *args, **kwargs) + finally: + db.close() + return wrapper + + class ClientConfigManager: """Manages client configurations using penguin-dal.""" @@ -187,9 +206,9 @@ def _extract_cn_from_subject(self, subject_dn: str) -> Optional[str]: pass return None - def _initialize_default_roles(self) -> None: + @_with_db + def _initialize_default_roles(self, db: DB) -> None: """Create default configuration roles if they don't exist.""" - db = self._get_db() try: # Define default roles default_roles = [ @@ -222,17 +241,16 @@ def _initialize_default_roles(self) -> None: db.commit() except Exception as e: logger.warning(f"Failed to initialize default roles: {e}") - finally: - db.close() + @_with_db def create_deployment_domain( self, + db: DB, name: str, description: str = "", created_by: str = "", ) -> Dict[str, Any]: """Create a new deployment domain with JWT token.""" - db = self._get_db() try: # Check for duplicate existing = db(db.deployment_domain.name == name).select().first() @@ -263,12 +281,10 @@ def create_deployment_domain( except Exception as e: logger.error(f"Failed to create deployment domain: {e}") return {"success": False, "error": str(e)} - finally: - db.close() - def rollover_domain_jwt(self, domain_id: int, admin_user: str = "") -> Dict[str, Any]: + @_with_db + def rollover_domain_jwt(self, db: DB, domain_id: int, admin_user: str = "") -> Dict[str, Any]: """Generate new JWT token for deployment domain.""" - db = self._get_db() try: domain = db(db.deployment_domain.id == domain_id).select().first() if not domain: @@ -294,11 +310,11 @@ def rollover_domain_jwt(self, domain_id: int, admin_user: str = "") -> Dict[str, except Exception as e: logger.error(f"Failed to rollover JWT: {e}") return {"success": False, "error": str(e)} - finally: - db.close() + @_with_db def create_client_config( self, + db: DB, name: str, domain_id: int, config_data: Dict[str, Any], @@ -306,7 +322,6 @@ def create_client_config( created_by: str = "", ) -> Dict[str, Any]: """Create a new client configuration.""" - db = self._get_db() try: # Validate config data if not self._validate_config_data(config_data): @@ -339,18 +354,17 @@ def create_client_config( except Exception as e: logger.error(f"Failed to create client config: {e}") return {"success": False, "error": str(e)} - finally: - db.close() + @_with_db def update_client_config( self, + db: DB, config_id: int, config_data: Dict[str, Any], description: str = "", changed_by: str = "", ) -> Dict[str, Any]: """Update an existing client configuration.""" - db = self._get_db() try: config = db(db.client_config.id == config_id).select().first() if not config: @@ -386,12 +400,10 @@ def update_client_config( except Exception as e: logger.error(f"Failed to update client config: {e}") return {"success": False, "error": str(e)} - finally: - db.close() - def _verify_domain_jwt(self, jwt_token: str) -> Optional[Dict[str, Any]]: + @_with_db + def _verify_domain_jwt(self, db: DB, jwt_token: str) -> Optional[Dict[str, Any]]: """Verify domain JWT token and return domain info.""" - db = self._get_db() try: # Verify signature with the public key. Expiry (`exp`) is enforced # natively by PyJWT; iss/aud are validated. HS256/none are rejected. @@ -428,8 +440,6 @@ def _verify_domain_jwt(self, jwt_token: str) -> Optional[Dict[str, Any]]: logger.warning("Invalid JWT token used for domain verification") except Exception as e: logger.error(f"JWT verification failed: {e}") - finally: - db.close() return None @@ -471,8 +481,10 @@ def _verify_user_token( logger.error(f"User token verification failed: {e}") return {"valid": False, "reason": "Token verification error"} + @_with_db def register_client( self, + db: DB, client_id: str, domain_jwt: str, hostname: str, @@ -483,8 +495,6 @@ def register_client( client_cert_subject: Optional[str] = None, ) -> Dict[str, Any]: """Register a new client instance.""" - db = self._get_db() - try: # Verify user auth if provided if user_token: @@ -555,19 +565,17 @@ def register_client( except Exception as e: logger.error(f"Failed to register client: {e}") return {"success": False, "error": str(e)} - finally: - db.close() + @_with_db def pull_client_config( self, + db: DB, client_id: str, domain_jwt: str, user_token: Optional[str] = None, client_cert_subject: Optional[str] = None, ) -> Dict[str, Any]: """Pull configuration for a client.""" - db = self._get_db() - try: # Verify user auth if provided if user_token: @@ -633,18 +641,16 @@ def pull_client_config( except Exception as e: logger.error(f"Failed to pull client config: {e}") return {"success": False, "error": str(e)} - finally: - db.close() + @_with_db def assign_config_to_client( self, + db: DB, client_id: str, config_id: int, assigned_by: str = "", ) -> Dict[str, Any]: """Assign a specific configuration to a client.""" - db = self._get_db() - try: client = db(db.client_instance.client_id == client_id).select().first() if not client: @@ -675,13 +681,10 @@ def assign_config_to_client( except Exception as e: logger.error(f"Failed to assign config to client: {e}") return {"success": False, "error": str(e)} - finally: - db.close() - def get_domain_clients(self, domain_id: int) -> List[Dict[str, Any]]: + @_with_db + def get_domain_clients(self, db: DB, domain_id: int) -> List[Dict[str, Any]]: """Get all clients in a deployment domain.""" - db = self._get_db() - try: clients = db(db.client_instance.domain_id == domain_id).select() @@ -717,13 +720,10 @@ def get_domain_clients(self, domain_id: int) -> List[Dict[str, Any]]: except Exception as e: logger.error(f"Failed to get domain clients: {e}") return [] - finally: - db.close() - def get_client_stats(self) -> Dict[str, Any]: + @_with_db + def get_client_stats(self, db: DB) -> Dict[str, Any]: """Get client configuration statistics.""" - db = self._get_db() - try: # Domain stats all_domains = db(db.deployment_domain.id > 0).count() # All deployment domains @@ -763,13 +763,10 @@ def get_client_stats(self) -> Dict[str, Any]: except Exception as e: logger.error(f"Failed to get client stats: {e}") return {} - finally: - db.close() - def cleanup_inactive_clients(self, inactive_days: int = 30) -> int: + @_with_db + def cleanup_inactive_clients(self, db: DB, inactive_days: int = 30) -> int: """Remove clients that haven't checked in for specified days.""" - db = self._get_db() - try: cutoff_time = datetime.now() - timedelta(days=inactive_days) @@ -786,5 +783,3 @@ def cleanup_inactive_clients(self, inactive_days: int = 30) -> int: except Exception as e: logger.error(f"Failed to cleanup inactive clients: {e}") return 0 - finally: - db.close() diff --git a/manager/backend/app/utils/crypto.py b/manager/backend/app/utils/crypto.py new file mode 100644 index 00000000..9ea9cbae --- /dev/null +++ b/manager/backend/app/utils/crypto.py @@ -0,0 +1,32 @@ +"""Shared crypto helpers for the manager service.""" + +from __future__ import annotations + +from typing import Tuple + + +def generate_ephemeral_es256_keypair() -> Tuple[str, str]: + """Generate an in-process ES256 (NIST P-256) keypair as PEM strings. + + Returns: + (private_pem, public_pem) + + Used only where no configured keypair is supplied (dev/test/standalone). + Production always supplies JWT_PRIVATE_KEY/JWT_PUBLIC_KEY — ProductionConfig + fails fast without them. + """ + 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 diff --git a/manager/backend/tests/conftest.py b/manager/backend/tests/conftest.py index fdb7cc5a..a9fb60db 100644 --- a/manager/backend/tests/conftest.py +++ b/manager/backend/tests/conftest.py @@ -13,9 +13,6 @@ import pytest import jwt from sqlalchemy import create_engine, text -from cryptography.hazmat.primitives.asymmetric import ec -from cryptography.hazmat.primitives import serialization -from cryptography.hazmat.backends import default_backend # Ensure app package is importable sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) @@ -95,24 +92,10 @@ def clean_tables(db): @pytest.fixture(scope="session") def jwt_keypair(): - """Generate an ephemeral ES256 keypair for testing.""" - # Generate EC private key (P-256) - 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') - - # Extract public key - public_key = private_key.public_key() - public_pem = public_key.public_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PublicFormat.SubjectPublicKeyInfo - ).decode('utf-8') + """Generate an ephemeral ES256 keypair for testing (shared helper).""" + from app.utils.crypto import generate_ephemeral_es256_keypair + private_pem, public_pem = generate_ephemeral_es256_keypair() return { 'private': private_pem, 'public': public_pem diff --git a/manager/backend/tests/test_client_config_domain_jwt.py b/manager/backend/tests/test_client_config_domain_jwt.py index e11b2d5b..52d93977 100644 --- a/manager/backend/tests/test_client_config_domain_jwt.py +++ b/manager/backend/tests/test_client_config_domain_jwt.py @@ -12,25 +12,9 @@ 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 +from app.utils.crypto import generate_ephemeral_es256_keypair as _gen_es256_pem class _FakeField: diff --git a/tests/test_installer.py b/tests/test_installer.py index 66462611..e220cf22 100644 --- a/tests/test_installer.py +++ b/tests/test_installer.py @@ -27,8 +27,15 @@ def installer(self): def test_check_admin_windows(self): """Test admin check on Windows""" + import ctypes + from unittest.mock import MagicMock + + # ctypes.windll only exists on Windows — attach a mock (create=True) + # so the Windows code path is exercisable on Linux/macOS CI. + windll = MagicMock() + windll.shell32.IsUserAnAdmin.return_value = 1 with patch('platform.system', return_value='Windows'): - with patch('ctypes.windll.shell32.IsUserAnAdmin', return_value=1): + with patch.object(ctypes, 'windll', windll, create=True): installer = SquawkInstaller() assert installer.is_admin is True