fix: address CodeQL security findings

Author: suguanYang

README.md

@@ -83,6 +83,10 @@ uv run --python 3.11 python -m alembic upgrade heads
 uv run --python 3.11 python scripts/init_user.py --email you@example.com
 ```
 
+Pass `--api-key-output-file ./standalone-api-key.txt` if you need the generated
+plaintext key written to a local `0600` file. The default console output only
+reports that the credential was created.
+
 If you plan to use the dashboard, start the combined self-hosted stack and
 register through the dashboard instead of using `scripts/init_user.py`.
 

apps/api/alembic/versions/f6a7b8c9d0e1_key_api_key_hashes.py

@@ -0,0 +1,31 @@
+"""key api key hashes
+
+Revision ID: f6a7b8c9d0e1
+Revises: e5f6a7b8c9d0
+Create Date: 2026-05-01 05:55:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+
+revision: str = "f6a7b8c9d0e1"
+down_revision: Union[str, Sequence[str], None] = "e5f6a7b8c9d0"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.execute("UPDATE api_keys SET is_active = false")
+    op.add_column(
+        "api_keys",
+        sa.Column("hash_version", sa.String(length=16), nullable=False, server_default="hmac-v1"),
+    )
+    op.alter_column("api_keys", "hash_version", server_default=None)
+
+
+def downgrade() -> None:
+    op.drop_column("api_keys", "hash_version")

apps/api/app/api/v1/health.py

@@ -17,13 +17,23 @@
 @router.get("/database/health")
 async def check_database_health():
     """Check database health status."""
-    return await get_database_health()
+    health = await get_database_health()
+    if "error" in health:
+        return {
+            "status": health.get("status", "unhealthy"),
+            "error": "Database health check failed",
+            "last_check": health.get("last_check"),
+        }
+    return health
 
 
 @router.get("/database/info")
 async def get_database_information():
     """Return database connection information."""
-    return await get_database_info()
+    info = await get_database_info()
+    if "error" in info:
+        return {"error": "Database information unavailable"}
+    return info
 
 
 @router.get("/database/performance")

apps/api/app/api/v1/routes/s3_events.py

@@ -5,9 +5,11 @@
 import base64
 import json
 import os
+import socket
 from typing import Any, Dict
 
 import aiohttp
+from aiohttp.abc import AbstractResolver
 from app.repositories.job_repository import JobRepository
 from app.services.knowledge.kb_orchestrator import KBOrchestrator
 from app.services.state_machine import JobStateMachine
@@ -19,9 +21,13 @@
 from shared.core.state_machine.states import JobStatus
 from shared.models.schemas.oss_event import OSSEvent
 from shared.models.schemas.s3_event import S3Event
+from shared.services.webhook.validator import validate_webhook_url_async
+from shared.utils.url_security import SafePublicHTTPURL
 
 router = APIRouter(tags=["Internal"])
 
+SNS_SUBSCRIPTION_TIMEOUT_SECONDS = 10
+
 
 def verify_sns_signature(request_body: bytes, signature: str, message: str) -> bool:
     """
@@ -207,22 +213,7 @@ async def handle_sns_event(body: bytes):
             if subscribe_url:
                 logger.info(f"SNS subscription confirmation URL: {subscribe_url}")
                 # Visit the URL to confirm the subscription.
-                try:
-                    async with aiohttp.ClientSession() as session:
-                        async with session.get(subscribe_url) as response:
-                            if response.status == 200:
-                                logger.info("SNS subscription confirmed successfully")
-                                return {"message": "SNS subscription confirmed"}
-                            else:
-                                logger.error(
-                                    f"SNS subscription confirmation failed, status={response.status}"
-                                )
-                                return {
-                                    "message": "SNS subscription confirmation failed"
-                                }
-                except Exception as e:
-                    logger.error(f"Failed to reach the SNS confirmation URL: {e}")
-                    return {"message": "SNS subscription confirmation failed"}
+                return await confirm_sns_subscription(subscribe_url)
             else:
                 logger.warning(
                     "SNS subscription confirmation did not include SubscribeURL"
@@ -274,6 +265,80 @@ async def handle_sns_event(body: bytes):
         raise
 
 
+async def confirm_sns_subscription(subscribe_url: str) -> dict[str, str]:
+    """Confirm an SNS subscription after SSRF validation and IP pinning."""
+    validation = await validate_webhook_url_async(subscribe_url)
+    if not validation.is_valid:
+        logger.warning(
+            f"SNS subscription confirmation URL failed validation: {validation.error_message}"
+        )
+        return {"message": "SNS subscription confirmation failed"}
+
+    if not validation.validated_ip:
+        logger.warning("SNS subscription confirmation URL validation returned no IP")
+        return {"message": "SNS subscription confirmation failed"}
+
+    try:
+        validated_subscribe_url = SafePublicHTTPURL(subscribe_url)
+        connector = aiohttp.TCPConnector(
+            resolver=_PinnedSNSResolver(validation.validated_ip),
+        )
+        timeout = aiohttp.ClientTimeout(total=SNS_SUBSCRIPTION_TIMEOUT_SECONDS)
+        async with aiohttp.ClientSession(
+            connector=connector,
+            timeout=timeout,
+        ) as session:
+            async with session.get(
+                validated_subscribe_url,
+                allow_redirects=False,
+            ) as response:
+                if response.status == 200:
+                    logger.info("SNS subscription confirmed successfully")
+                    return {"message": "SNS subscription confirmed"}
+
+                if 300 <= response.status < 400:
+                    logger.warning(
+                        f"SNS subscription confirmation redirect blocked, status={response.status}"
+                    )
+                else:
+                    logger.error(
+                        f"SNS subscription confirmation failed, status={response.status}"
+                    )
+                return {"message": "SNS subscription confirmation failed"}
+    except Exception as e:
+        logger.error(f"Failed to reach the SNS confirmation URL: {e}")
+        return {"message": "SNS subscription confirmation failed"}
+
+
+class _PinnedSNSResolver(AbstractResolver):
+    """Resolver that pins SNS confirmation to a pre-validated public IP."""
+
+    def __init__(self, pinned_ip: str) -> None:
+        self.pinned_ip = pinned_ip
+
+    async def resolve(
+        self,
+        host: str,
+        port: int = 0,
+        family: int = socket.AF_INET,
+    ) -> list[dict[str, Any]]:
+        parsed_ip = self.pinned_ip
+        pinned_family = socket.AF_INET6 if ":" in parsed_ip else socket.AF_INET
+        return [
+            {
+                "hostname": host,
+                "host": parsed_ip,
+                "port": port,
+                "family": pinned_family,
+                "proto": 0,
+                "flags": socket.AI_NUMERICHOST,
+            }
+        ]
+
+    async def close(self) -> None:
+        pass
+
+
 async def handle_minio_event(body: bytes, auth_token: str):
     """
     Handle a MinIO webhook event.

apps/api/app/core/dependencies.py

@@ -1,4 +1,3 @@
-import hashlib
 import threading
 from datetime import timedelta
 from fnmatch import fnmatch
@@ -18,6 +17,7 @@
     AuthException,
     PermissionDeniedException,
 )
+from shared.utils.api_key_hashing import hash_api_key
 
 # Standard JWKS endpoint path (fixed, following OpenID Connect convention)
 JWKS_ENDPOINT_PATH = "/api/auth/jwks"
@@ -186,7 +186,7 @@ async def get_current_user_id(
     # Mode 1: API Key verification (for external clients)
     if token.startswith("sk_"):
         # Check identity cache first — skip DB on cache hit
-        api_key_hash = hashlib.sha256(token.encode()).hexdigest()
+        api_key_hash = hash_api_key(token)
         try:
             cached = await identity_cache.get_cached_identity(
                 redis_pool_manager.get_redis_service(),

apps/api/app/services/auth/api_key_service.py

@@ -1,7 +1,6 @@
 """API key management service."""
 
 import asyncio
-import hashlib
 import uuid
 from dataclasses import dataclass
 from datetime import datetime
@@ -23,6 +22,7 @@
 )
 from shared.models.database.api_key import APIKey
 from shared.models.database.user_balance import UserBalance
+from shared.utils.api_key_hashing import hash_api_key
 
 _DEFAULT_USER_TIER: str = "free"
 
@@ -85,7 +85,7 @@ async def create_api_key(
 
         # 3. Generate a secure API key (sk_ + a 32-char UUID without hyphens).
         api_key = f"sk_{str(uuid.uuid4()).replace('-', '')}"
-        key_hash = hashlib.sha256(api_key.encode()).hexdigest()
+        key_hash = hash_api_key(api_key)
         key_mask = self._mask_api_key(api_key)
 
         # 4. Store it in the database.
@@ -116,7 +116,7 @@ async def validate_api_key_identity(
         api_key: str,
     ) -> Optional[APIKeyIdentity]:
         """Validate API key and return the authenticated identity."""
-        key_hash = hashlib.sha256(api_key.encode()).hexdigest()
+        key_hash = hash_api_key(api_key)
         api_key_record = await self.repository.get_by_key_hash(session, key_hash)
 
         if not api_key_record or not api_key_record.is_valid():
@@ -236,7 +236,7 @@ async def regenerate_api_key(
 
         # 2. Generate a new API key (sk_ + a 32-char UUID without hyphens).
         new_api_key = f"sk_{str(uuid.uuid4()).replace('-', '')}"
-        new_key_hash = hashlib.sha256(new_api_key.encode()).hexdigest()
+        new_key_hash = hash_api_key(new_api_key)
         new_key_mask = self._mask_api_key(new_api_key)
 
         # 3. Update the database record.
@@ -267,7 +267,7 @@ async def check_module_permission(
         self, session: AsyncSession, api_key: str, module: str
     ) -> bool:
         """Check whether an API key can access the requested module."""
-        key_hash = hashlib.sha256(api_key.encode()).hexdigest()
+        key_hash = hash_api_key(api_key)
         api_key_record = await self.repository.get_by_key_hash(session, key_hash)
 
         if not api_key_record or not api_key_record.is_valid():

apps/api/app/services/guest/guest_registration_service.py

@@ -1,6 +1,7 @@
 """Guest registration business logic."""
 
 import hashlib
+import uuid
 from datetime import datetime
 from typing import NoReturn
 from uuid import uuid4
@@ -25,6 +26,7 @@
     GuestRegisterResponse,
 )
 from shared.services.billing.credits_service import CreditsService
+from shared.utils.api_key_hashing import hash_api_key
 
 _GUEST_TIER: str = "guest"
 _GUEST_KEY_NAME_PREFIX: str = "guest-device"
@@ -153,13 +155,10 @@ async def _create_api_key_without_commit(
         This avoids the internal commit inside APIKeyService.create_api_key()
         which would make the key durable before the device row is inserted.
         """
-        import hashlib
-        import uuid
-
         from shared.models.database.api_key import APIKey
 
         api_key = f"sk_{str(uuid.uuid4()).replace('-', '')}"
-        key_hash = hashlib.sha256(api_key.encode()).hexdigest()
+        key_hash = hash_api_key(api_key)
         key_mask = self._api_key_service._mask_api_key(api_key)
 
         api_key_record = APIKey(
@@ -264,13 +263,11 @@ def _raise_existing_device_conflict(cls, device_id: str) -> NoReturn:
     @staticmethod
     async def _resolve_api_key_id(session: AsyncSession, api_key: str) -> str | None:
         """Resolve the DB id for a just-created API key by its hash."""
-        import hashlib
-
         from sqlalchemy import select
 
         from shared.models.database.api_key import APIKey
 
-        key_hash = hashlib.sha256(api_key.encode()).hexdigest()
+        key_hash = hash_api_key(api_key)
         result = await session.execute(
             select(APIKey.id).where(APIKey.key_hash == key_hash).limit(1)
         )

apps/api/app/services/rate_limit/dependencies.py

@@ -14,7 +14,6 @@
 only when billing is enabled.
 """
 
-import hashlib
 import math
 from datetime import datetime, timezone
 from typing import AsyncGenerator
@@ -42,6 +41,7 @@
 from shared.core.logging import log_context
 from shared.core.state_machine.states import JobStatus
 from shared.models.database.api_key import APIKey
+from shared.utils.api_key_hashing import hash_api_key
 from shared.models.database.job import Job
 from shared.models.database.user_balance import UserBalance
 
@@ -175,7 +175,7 @@ async def with_current_user(
             api_key_hash = None
             is_api_key_auth = isinstance(token, str) and token.startswith("sk_")
             if token is not None and is_api_key_auth:
-                api_key_hash = hashlib.sha256(token.encode()).hexdigest()
+                api_key_hash = hash_api_key(token)
             if is_api_key_auth and api_key_hash:
                 try:
                     ttl_seconds = await _resolve_apikey_cache_ttl_seconds(api_key_hash)
@@ -197,7 +197,7 @@ async def with_current_user(
         api_key_hash = None
         is_api_key_auth = isinstance(token, str) and token.startswith("sk_")
         if token is not None and is_api_key_auth:
-            api_key_hash = hashlib.sha256(token.encode()).hexdigest()
+            api_key_hash = hash_api_key(token)
         cache_key: str = (
             identity_cache._apikey_key(api_key_hash)
             if is_api_key_auth and api_key_hash

apps/api/scripts/bootstrap_local_dev.py

@@ -45,12 +45,11 @@ async def _run(mode: str) -> int:
 
 
 def _print_profile() -> None:
-    profile = LocalDevelopmentBootstrapService.get_local_developer_profile()
-    print(f"user_id={profile['user_id']}")
-    print(f"name={profile['name']}")
-    print(f"email={profile['email']}")
-    print(f"tier={profile['tier']}")
-    print(f"api_key={profile['api_key']}")
+    print("user_id=local-dev-user")
+    print("name=Local Development User")
+    print("email=local-dev-user@knowhere.local")
+    print("tier=tier_5")
+    print("local_developer_key_seeded=true")
 
 
 def main() -> int: