Claude-coauthored key revocation trusted a target-key signature without checking that the authenticated requester owned the key, allowing another client with a valid signature to revoke it.

+Endpoints:+- POST /api/v1/keys/register - Register new client (no auth)+- POST /api/v1/keys - Upload key (auth required)+- GET /api/v1/keys/search - Search key (public)+- POST /api/v1/keys/{fingerprint}/revoke - Revoke key (auth required)+"""++import logging+from typing import Optional++from fastapi import APIRouter, Depends, Query, Request, Security, status+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer+from sqlalchemy.ext.asyncio import AsyncSession++from ...core.database import get_db+from .auth import get_keyserver_auth+from .schemas import (+    ErrorResponse,+    KeyBundleSchema,+    KeySearchResponse,+    KeyUploadResponse,+    RegisterResponse,+    RevocationRequest,+    RevocationResponse,+)+from .service import KeyserverService++logger = logging.getLogger(__name__)++router = APIRouter(prefix="/api/v1/keys", tags=["keyserver"])++security = HTTPBearer()+++# Dependency that lazily gets the auth instance+async def get_current_client(+    request: Request,+    credentials: HTTPAuthorizationCredentials = Security(security),+) -> str:+    """Get current authenticated client ID"""+    auth = get_keyserver_auth()+    payload = auth.verify_token(credentials.credentials)++    # Update last seen (fire and forget)+    try:+        await auth.update_last_seen(payload.sub)+    except Exception:+        pass++    return payload.sub++[email protected](+    "/register",+    response_model=RegisterResponse,+    status_code=status.HTTP_200_OK,+    summary="Register new keyserver client",+)+async def register():+    """+    Register a new Keyserver client.++    Returns a JWT token that can ONLY be used for Keyserver endpoints.+    The token includes an issuer claim that prevents cross-module usage.++    Returns:+        RegisterResponse: Client ID, JWT token, expiration+    """+    auth = get_keyserver_auth()+    return await auth.register_client()++[email protected](+    "",+    response_model=KeyUploadResponse,+    status_code=status.HTTP_200_OK,+    responses={+        400: {"model": ErrorResponse, "description": "Invalid bundle or verification failed"},+        401: {"model": ErrorResponse, "description": "Authentication required"},+        409: {"model": ErrorResponse, "description": "Key already exists"},+    },+    summary="Upload public key",+)+async def upload_key(+    bundle: KeyBundleSchema,

+Table prefix: ks_ (keyserver)+"""++import uuid+from datetime import datetime, timezone+from typing import Optional++from sqlalchemy import Boolean, Column, DateTime, Integer, String, Text+from sqlalchemy.dialects.postgresql import JSON, UUID++from ...core.database import Base+++class KSClient(Base):+    """+    Keyserver client (API token authenticated).++    Each client receives a unique JWT token for accessing keyserver endpoints.+    """++    __tablename__ = "ks_clients"++    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)+    client_id = Column(String(64), unique=True, nullable=False, index=True)+    created_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc))+    last_seen_at = Column(DateTime(timezone=True), nullable=True)+    client_metadata = Column(JSON, nullable=True)  # Optional client info++    def __repr__(self):+        return f"<KSClient(client_id={self.client_id})>"+++class KSKey(Base):+    """+    Public key storage.++    Stores post-quantum public keys with metadata and verification status.+    """++    __tablename__ = "ks_keys"++    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)+    fingerprint = Column(String(64), unique=True, nullable=False, index=True)+    name = Column(String(255), nullable=True, index=True)+    email = Column(String(255), nullable=True, index=True)+    bundle_json = Column(Text, nullable=False)  # Complete PublicKeyBundle JSON+    encryption_algorithm = Column(String(50), nullable=False)  # e.g., "ML-KEM-768"+    signing_algorithm = Column(String(50), nullable=False)  # e.g., "ML-DSA-65"+    revoked = Column(Boolean, nullable=False, default=False)+    revoked_at = Column(DateTime(timezone=True), nullable=True)+    revocation_reason = Column(Text, nullable=True)+    owner_client_id = Column(String(64), nullable=True)  # Optional: track uploader+    created_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc))+    updated_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))+    upload_count = Column(Integer, nullable=False, default=1)++    def __repr__(self):+        return f"<KSKey(fingerprint={self.fingerprint[:20]}..., name={self.name})>"+++class KSAccessLog(Base):+    """+    Access log for keyserver operations.++    Tracks key uploads, downloads, searches, and revocations.+    """++    __tablename__ = "ks_access_log"++    id = Column(Integer, primary_key=True, autoincrement=True)+    key_fingerprint = Column(String(64), nullable=False)+    action = Column(String(20), nullable=False)  # 'upload', 'download', 'search', 'revoke'+    client_id = Column(String(64), nullable=True)+    ip_address = Column(String(45), nullable=True)  # IPv4/IPv6+    timestamp = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc), index=True)++    def __repr__(self):+        return f"<KSAccessLog(action={self.action}, fingerprint={self.key_fingerprint[:20]}...)>"

+        This creates a new client registration on the keyserver and returns+        a JWT token that can be used for authenticated operations (upload/revoke).++        Args:+            server_url: Optional specific server URL. If None, uses first configured server.++        Returns:+            dict with keys: client_id, token, expires_at, token_type++        Raises:+            NetworkError: If network request fails+            ValueError: If no servers configured+        """+        if not self.config.enabled:+            raise ValueError("Keyserver plugin disabled")++        # Determine which server to register with+        if server_url is None:+            if not self.config.servers:+                raise ValueError("No keyservers configured")+            server_url = self.config.servers[0]++        register_url = f"{server_url}/api/v1/keys/register"++        try:+            # HTTP POST (no authentication required for registration)+            response = requests.post(+                register_url,+                timeout=(self.config.connect_timeout_seconds, self.config.read_timeout_seconds),+            )++            if response.status_code == 200:+                data = response.json()+                # Save JWT token to token file+                self.config.save_api_token(data["token"])+                logger.info(f"Registered with keyserver, client_id={data['client_id']}")+                return data+            else:+                raise NetworkError(+                    f"Registration failed with status {response.status_code}: {response.text}"+                )++        except requests.exceptions.Timeout:+            raise NetworkError("Request timeout")+        except requests.exceptions.ConnectionError as e:+            raise NetworkError(f"Connection failed: {e}")+        except requests.exceptions.RequestException as e:+            raise NetworkError(f"Request failed: {e}")+     def upload_key(self, bundle: PublicKeyBundle) -> bool:         """         Upload public key bundle to keyserver. -        SECURITY: Requires API token for authentication (Bearer token).+        SECURITY: Requires JWT token for authentication (Bearer token).          Args:             bundle: PublicKeyBundle to upload@@ -255,7 +308,7 @@ def upload_key(self, bundle: PublicKeyBundle) -> bool:             True if uploaded successfully, False otherwise          Raises:-            AuthenticationError: If API token not set or invalid+            AuthenticationError: If JWT token not set or invalid             NetworkError: If network request fails         """         if not self.config.enabled:@@ -266,11 +319,11 @@ def upload_key(self, bundle: PublicKeyBundle) -> bool:             logger.error("Uploads disabled in configuration")             return False -        # Get API token+        # Get JWT token         api_token = self.config.load_api_token()         if not api_token:             raise AuthenticationError(-                "API token not set. Use 'openssl-encrypt keyserver set-token' to set token."+                "JWT token not set. Use 'openssl-encrypt keyserver register' to register and obtain token."             )          # Verify bundle signature before uploading@@ -355,7 +408,7 @@ def revoke_key(self, fingerprint: str, signature: bytes) -> bool:

-Endpoints:-- POST /api/v1/keys/register - Register new client (no auth)-- POST /api/v1/keys - Upload key (auth required)-- GET /api/v1/keys/search - Search key (public)-- POST /api/v1/keys/{fingerprint}/revoke - Revoke key (auth required)-"""--import logging-from typing import Optional--from fastapi import APIRouter, Depends, Query, Request, Security, status-from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer-from sqlalchemy.ext.asyncio import AsyncSession-from slowapi import Limiter-from slowapi.util import get_remote_address--from ...core.database import get_db-from .auth import get_keyserver_auth-from .schemas import (-    ErrorResponse,-    KeyBundleSchema,-    KeySearchResponse,-    KeyUploadResponse,-    RegisterResponse,-    RevocationRequest,-    RevocationResponse,-)-from .service import KeyserverService--logger = logging.getLogger(__name__)--router = APIRouter(prefix="/api/v1/keys", tags=["keyserver"])--security = HTTPBearer()--# Initialize rate limiter-limiter = Limiter(key_func=get_remote_address)---# Dependency that lazily gets the auth instance-async def get_current_client(-    request: Request,-    credentials: HTTPAuthorizationCredentials = Security(security),-) -> str:-    """Get current authenticated client ID"""-    auth = get_keyserver_auth()-    payload = auth.verify_token(credentials.credentials)--    # Update last seen (fire and forget)-    try:-        await auth.update_last_seen(payload.sub)-    except Exception:-        pass--    return payload.sub--[email protected](-    "/register",-    response_model=RegisterResponse,-    status_code=status.HTTP_200_OK,-    summary="Register new keyserver client",-)[email protected]("10/hour")-async def register(request: Request):-    """-    Register a new Keyserver client.--    Returns a JWT token that can ONLY be used for Keyserver endpoints.-    The token includes an issuer claim that prevents cross-module usage.--    Returns:-        RegisterResponse: Client ID, JWT token, expiration-    """-    auth = get_keyserver_auth()-    return await auth.register_client()--[email protected](-    "",-    response_model=KeyUploadResponse,-    status_code=status.HTTP_200_OK,-    responses={-        400: {"model": ErrorResponse, "description": "Invalid bundle or verification failed"},-        401: {"model": ErrorResponse, "description": "Authentication required"},

-Table prefix: ks_ (keyserver)-"""--import uuid-from datetime import datetime, timezone-from typing import Optional--from sqlalchemy import Boolean, Column, DateTime, Integer, String, Text-from sqlalchemy.dialects.postgresql import JSON, UUID--from ...core.database import Base---class KSClient(Base):-    """-    Keyserver client (API token authenticated).--    Each client receives a unique JWT token for accessing keyserver endpoints.-    """--    __tablename__ = "ks_clients"--    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)-    client_id = Column(String(64), unique=True, nullable=False, index=True)-    created_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc))-    last_seen_at = Column(DateTime(timezone=True), nullable=True)-    client_metadata = Column(JSON, nullable=True)  # Optional client info--    def __repr__(self):-        return f"<KSClient(client_id={self.client_id})>"---class KSKey(Base):-    """-    Public key storage.--    Stores post-quantum public keys with metadata and verification status.-    """--    __tablename__ = "ks_keys"--    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)-    fingerprint = Column(String(100), unique=True, nullable=False, index=True)  # SHA-256 with colons: 95 chars-    name = Column(String(255), nullable=True, index=True)-    email = Column(String(255), nullable=True, index=True)-    bundle_json = Column(Text, nullable=False)  # Complete PublicKeyBundle JSON-    encryption_algorithm = Column(String(50), nullable=False)  # e.g., "ML-KEM-768"-    signing_algorithm = Column(String(50), nullable=False)  # e.g., "ML-DSA-65"-    revoked = Column(Boolean, nullable=False, default=False)-    revoked_at = Column(DateTime(timezone=True), nullable=True)-    revocation_reason = Column(Text, nullable=True)-    owner_client_id = Column(String(64), nullable=True)  # Optional: track uploader-    created_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc))-    updated_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))-    upload_count = Column(Integer, nullable=False, default=1)--    def __repr__(self):-        return f"<KSKey(fingerprint={self.fingerprint[:20]}..., name={self.name})>"---class KSAccessLog(Base):-    """-    Access log for keyserver operations.--    Tracks key uploads, downloads, searches, and revocations.-    """--    __tablename__ = "ks_access_log"--    id = Column(Integer, primary_key=True, autoincrement=True)-    key_fingerprint = Column(String(100), nullable=False)  # SHA-256 with colons: 95 chars-    action = Column(String(20), nullable=False)  # 'upload', 'download', 'search', 'revoke'-    client_id = Column(String(64), nullable=True)-    ip_address = Column(String(45), nullable=True)  # IPv4/IPv6-    timestamp = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc), index=True)--    def __repr__(self):-        return f"<KSAccessLog(action={self.action}, fingerprint={self.key_fingerprint[:20]}...)>"

-Tests the new refresh token functionality with sliding expiration,-token type validation, and security boundaries.-"""--import pytest-from datetime import datetime, timedelta, timezone-from unittest.mock import patch, MagicMock-import jwt--from openssl_encrypt_server.core.auth.token import TokenAuth, TokenConfig-from openssl_encrypt_server.modules.keyserver.models import KSClient-from fastapi import HTTPException--[email protected]-def token_config():-    """Create test token configuration"""-    return TokenConfig(-        secret="test_secret_key_for_testing_only",-        algorithm="HS256",-        access_token_expire_minutes=60,  # 1 hour-        refresh_token_expire_days=7,     # 7 days-        issuer="test-keyserver",-        enable_sliding_expiration=True-    )--[email protected]-def token_auth(token_config):-    """Create TokenAuth instance for testing"""-    return TokenAuth(token_config, KSClient)---class TestAccessTokenExpiry:-    """Tests for access token expiration"""--    def test_access_token_expires_after_1_hour(self, token_auth):-        """Access token should expire after 1 hour"""-        client_id = "test_client_123"--        # Create token-        access_token, expiry = token_auth.create_token(client_id, "access")--        # Verify expiry is approximately 1 hour from now-        now = datetime.now(timezone.utc)-        expected_expiry = now + timedelta(hours=1)--        # Allow 5 second tolerance for test execution time-        assert abs((expiry - expected_expiry).total_seconds()) < 5--        # Decode and verify expiry claim-        payload = jwt.decode(-            access_token,-            token_auth.secret,-            algorithms=[token_auth.algorithm],-            issuer=token_auth.issuer-        )--        exp_time = datetime.fromtimestamp(payload['exp'], tz=timezone.utc)-        assert abs((exp_time - expected_expiry).total_seconds()) < 5--    def test_expired_access_token_rejected(self, token_auth):-        """Expired access token should be rejected"""-        client_id = "test_client_123"--        # Create token with past expiry-        with patch('openssl_encrypt_server.core.auth.token.datetime') as mock_datetime:-            past_time = datetime.now(timezone.utc) - timedelta(hours=2)-            mock_datetime.now.return_value = past_time--            access_token, _ = token_auth.create_token(client_id, "access")--        # Try to verify expired token-        with pytest.raises(HTTPException) as exc_info:-            token_auth.verify_token(access_token)--        assert exc_info.value.status_code == 401-        assert "Token expired" in exc_info.value.detail---class TestRefreshTokenExpiry:-    """Tests for refresh token expiration"""--    def test_refresh_token_expires_after_7_days(self, token_auth):-        """Refresh token should expire after 7 days"""