refactor: de-dup JWT verification, DB lifecycle, and keypair generation

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]]:

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

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."""

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(