Merge pull request #73 from christianlouis/copilot/implement-database-backed-key-storage
Fix CodeQL clear-text logging alerts and improve startup branch coverage
This commit is contained in:
@@ -11,6 +11,12 @@ PROJECT_NAME="DMARQ"
|
|||||||
# NEVER use the default value in production!
|
# NEVER use the default value in production!
|
||||||
SECRET_KEY="CHANGE_THIS_TO_A_RANDOM_SECRET_IN_PRODUCTION"
|
SECRET_KEY="CHANGE_THIS_TO_A_RANDOM_SECRET_IN_PRODUCTION"
|
||||||
|
|
||||||
|
# Admin API Key for the X-API-Key header (optional)
|
||||||
|
# If set, this key is used instead of generating a random one at startup.
|
||||||
|
# Generate with: openssl rand -hex 32
|
||||||
|
# If not set, a random key is generated each restart (key length logged only).
|
||||||
|
# ADMIN_API_KEY="your_admin_api_key_here"
|
||||||
|
|
||||||
# Environment (development/production)
|
# Environment (development/production)
|
||||||
# Affects HSTS and other security settings
|
# Affects HSTS and other security settings
|
||||||
ENVIRONMENT="development"
|
ENVIRONMENT="development"
|
||||||
|
|||||||
@@ -49,6 +49,24 @@ class Settings(BaseSettings):
|
|||||||
CLOUDFLARE_API_TOKEN: Optional[str] = None
|
CLOUDFLARE_API_TOKEN: Optional[str] = None
|
||||||
CLOUDFLARE_ZONE_ID: Optional[str] = None
|
CLOUDFLARE_ZONE_ID: Optional[str] = None
|
||||||
|
|
||||||
|
# Admin API Key (optional)
|
||||||
|
# If set, this key is used directly instead of generating a random one at startup.
|
||||||
|
# Use: openssl rand -hex 32
|
||||||
|
ADMIN_API_KEY: Optional[str] = None
|
||||||
|
|
||||||
|
@validator("ADMIN_API_KEY", pre=True, always=True)
|
||||||
|
@classmethod
|
||||||
|
def validate_admin_api_key(cls, v: Optional[str]) -> Optional[str]: # pylint: disable=no-self-argument
|
||||||
|
"""Warn if ADMIN_API_KEY is set but too short."""
|
||||||
|
if v is not None and len(v) < 32:
|
||||||
|
logger.warning(
|
||||||
|
"ADMIN_API_KEY is too short (%s characters). "
|
||||||
|
"Recommended minimum is 32 characters for security. "
|
||||||
|
"Generate a strong key with: openssl rand -hex 32",
|
||||||
|
len(v),
|
||||||
|
)
|
||||||
|
return v or None
|
||||||
|
|
||||||
@validator("SECRET_KEY", pre=True, always=True)
|
@validator("SECRET_KEY", pre=True, always=True)
|
||||||
def validate_secret_key(cls, v: Optional[str]) -> str: # pylint: disable=no-self-argument
|
def validate_secret_key(cls, v: Optional[str]) -> str: # pylint: disable=no-self-argument
|
||||||
"""Validate and generate SECRET_KEY if not provided."""
|
"""Validate and generate SECRET_KEY if not provided."""
|
||||||
|
|||||||
@@ -76,7 +76,7 @@ def add_api_key(api_key: str) -> bool:
|
|||||||
if api_key in _api_keys:
|
if api_key in _api_keys:
|
||||||
return False
|
return False
|
||||||
_api_keys.add(api_key)
|
_api_keys.add(api_key)
|
||||||
logger.info("API key added (ends with: ...%s)", api_key[-8:])
|
logger.info("API key added (length: %d chars)", len(api_key))
|
||||||
return True
|
return True
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
+21
-15
@@ -295,21 +295,27 @@ def create_app() -> FastAPI:
|
|||||||
# Ensure all tables exist (no-op if already present)
|
# Ensure all tables exist (no-op if already present)
|
||||||
Base.metadata.create_all(bind=engine)
|
Base.metadata.create_all(bind=engine)
|
||||||
|
|
||||||
# Generate and provide admin API key
|
# Load or generate the admin API key
|
||||||
api_key = generate_api_key()
|
if settings.ADMIN_API_KEY:
|
||||||
add_api_key(api_key)
|
api_key = settings.ADMIN_API_KEY
|
||||||
|
add_api_key(api_key)
|
||||||
# Security: Log only last 8 characters for reference
|
logger.info(
|
||||||
logger.warning(
|
"Admin API key loaded from ADMIN_API_KEY environment variable "
|
||||||
"%s\nIMPORTANT: Admin API Key Generated\n"
|
"(length: %d chars).",
|
||||||
"API Key (last 8 chars): ...%s\n"
|
len(api_key),
|
||||||
"Full key stored securely in memory.\n"
|
)
|
||||||
"For production, retrieve the key through secure configuration management.\n"
|
else:
|
||||||
"Use this key in the X-API-Key header for admin endpoints.\n%s",
|
api_key = generate_api_key()
|
||||||
"=" * 80,
|
add_api_key(api_key)
|
||||||
api_key[-8:],
|
logger.warning(
|
||||||
"=" * 80,
|
"%s\nIMPORTANT: Admin API Key Generated\n"
|
||||||
)
|
"Key length: %d chars. Full key stored securely in memory.\n"
|
||||||
|
"Set ADMIN_API_KEY in your environment to use a fixed key across restarts.\n"
|
||||||
|
"Use this key in the X-API-Key header for admin endpoints.\n%s",
|
||||||
|
"=" * 80,
|
||||||
|
len(api_key),
|
||||||
|
"=" * 80,
|
||||||
|
)
|
||||||
|
|
||||||
# One-time migration: if IMAP_* env vars are set and no mail sources exist,
|
# One-time migration: if IMAP_* env vars are set and no mail sources exist,
|
||||||
# create an initial MailSource from those settings so existing deployments
|
# create an initial MailSource from those settings so existing deployments
|
||||||
|
|||||||
@@ -138,3 +138,35 @@ class TestEnsureSqliteDir:
|
|||||||
"""Default DATABASE_URL places the SQLite file inside a data/ subdirectory."""
|
"""Default DATABASE_URL places the SQLite file inside a data/ subdirectory."""
|
||||||
settings = Settings()
|
settings = Settings()
|
||||||
assert settings.DATABASE_URL.endswith("data/dmarq.db")
|
assert settings.DATABASE_URL.endswith("data/dmarq.db")
|
||||||
|
|
||||||
|
|
||||||
|
class TestAdminApiKeySetting:
|
||||||
|
"""Tests for the ADMIN_API_KEY settings field."""
|
||||||
|
|
||||||
|
def test_admin_api_key_defaults_to_none(self):
|
||||||
|
"""ADMIN_API_KEY is None when not set."""
|
||||||
|
settings = Settings()
|
||||||
|
assert settings.ADMIN_API_KEY is None
|
||||||
|
|
||||||
|
def test_admin_api_key_reads_from_env(self, monkeypatch):
|
||||||
|
"""ADMIN_API_KEY is read from the environment variable."""
|
||||||
|
monkeypatch.setenv("ADMIN_API_KEY", "mytestapikey1234")
|
||||||
|
settings = Settings()
|
||||||
|
assert settings.ADMIN_API_KEY == "mytestapikey1234"
|
||||||
|
|
||||||
|
def test_admin_api_key_warns_when_short(self, monkeypatch, caplog):
|
||||||
|
"""A warning is logged when ADMIN_API_KEY is shorter than 32 characters."""
|
||||||
|
import logging
|
||||||
|
|
||||||
|
monkeypatch.setenv("ADMIN_API_KEY", "short")
|
||||||
|
with caplog.at_level(logging.WARNING, logger="app.core.config"):
|
||||||
|
settings = Settings()
|
||||||
|
assert settings.ADMIN_API_KEY == "short"
|
||||||
|
assert any("too short" in record.message for record in caplog.records)
|
||||||
|
|
||||||
|
def test_admin_api_key_accepts_long_key(self, monkeypatch):
|
||||||
|
"""A 64-char hex key (openssl rand -hex 32 output) is accepted without warnings."""
|
||||||
|
long_key = "a" * 64
|
||||||
|
monkeypatch.setenv("ADMIN_API_KEY", long_key)
|
||||||
|
settings = Settings()
|
||||||
|
assert settings.ADMIN_API_KEY == long_key
|
||||||
|
|||||||
@@ -5,9 +5,11 @@ Covers API key management, domain validation, file upload limits, and XML parsin
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
from fastapi.testclient import TestClient
|
||||||
|
|
||||||
import app.services.dmarc_parser as parser_module
|
import app.services.dmarc_parser as parser_module
|
||||||
from app.core.security import add_api_key, generate_api_key, verify_api_key
|
from app.core.security import add_api_key, generate_api_key, verify_api_key
|
||||||
|
from app.main import create_app
|
||||||
from app.services.dmarc_parser import DMARCParser
|
from app.services.dmarc_parser import DMARCParser
|
||||||
from app.utils.domain_validator import validate_domain, validate_domain_config
|
from app.utils.domain_validator import validate_domain, validate_domain_config
|
||||||
|
|
||||||
@@ -151,3 +153,42 @@ class TestXMLParsingSecurity:
|
|||||||
assert "root:" not in org_name and "/bin" not in org_name
|
assert "root:" not in org_name and "/bin" not in org_name
|
||||||
except Exception: # pylint: disable=broad-exception-caught
|
except Exception: # pylint: disable=broad-exception-caught
|
||||||
pass # Expected – defusedxml blocks DTD processing
|
pass # Expected – defusedxml blocks DTD processing
|
||||||
|
|
||||||
|
|
||||||
|
class TestAdminApiKeyStartup:
|
||||||
|
"""Test admin API key loading during the application startup event."""
|
||||||
|
|
||||||
|
def test_startup_uses_env_api_key(self, monkeypatch):
|
||||||
|
"""When ADMIN_API_KEY is configured, startup should register it directly."""
|
||||||
|
import app.core.security as sec_module
|
||||||
|
import app.main as main_module
|
||||||
|
|
||||||
|
test_key = "a" * 64
|
||||||
|
monkeypatch.setattr(main_module.settings, "ADMIN_API_KEY", test_key)
|
||||||
|
|
||||||
|
saved_keys = set(sec_module._api_keys)
|
||||||
|
sec_module._api_keys.clear()
|
||||||
|
try:
|
||||||
|
application = create_app()
|
||||||
|
with TestClient(application):
|
||||||
|
assert sec_module.verify_api_key(test_key)
|
||||||
|
finally:
|
||||||
|
sec_module._api_keys.clear()
|
||||||
|
sec_module._api_keys.update(saved_keys)
|
||||||
|
|
||||||
|
def test_startup_generates_key_when_no_env(self, monkeypatch):
|
||||||
|
"""When ADMIN_API_KEY is not set, startup should generate a random key."""
|
||||||
|
import app.core.security as sec_module
|
||||||
|
import app.main as main_module
|
||||||
|
|
||||||
|
monkeypatch.setattr(main_module.settings, "ADMIN_API_KEY", None)
|
||||||
|
|
||||||
|
saved_keys = set(sec_module._api_keys)
|
||||||
|
sec_module._api_keys.clear()
|
||||||
|
try:
|
||||||
|
application = create_app()
|
||||||
|
with TestClient(application):
|
||||||
|
assert len(sec_module._api_keys) == 1
|
||||||
|
finally:
|
||||||
|
sec_module._api_keys.clear()
|
||||||
|
sec_module._api_keys.update(saved_keys)
|
||||||
|
|||||||
Reference in New Issue
Block a user