feat(auth): implement API key authentication and management
Introduce a new API key authentication mechanism to support service-to-service communication (e.g., Airflow, CI/CD) without browser-based JWT login. - Add `APIKey` model and `APIKeyPrincipal` dependency for RBAC. - Implement `require_api_key_or_jwt` dependency with environment scoping. - Add admin CRUD endpoints for API key lifecycle management. - Add API key management UI in System Settings. - Update maintenance API to support explicit `environment_id` and API key auth. - Add i18n support for API keys and connection settings. - Include Python and Bash usage examples in the `examples/` directory. - Update technical specifications and documentation.
This commit is contained in:
122
backend/tests/test_api_key_model.py
Normal file
122
backend/tests/test_api_key_model.py
Normal file
@@ -0,0 +1,122 @@
|
||||
# #region TestAPIKeyModel [C:2] [TYPE Module] [SEMANTICS test, api_key, model]
|
||||
# @BRIEF Contract tests for the APIKey SQLAlchemy model — creation, hash storage, and field constraints.
|
||||
# @RELATION BINDS_TO -> [APIKeyModel]
|
||||
# @TEST_CONTRACT: APIKey model stores SHA-256 hash and prefix; raw key never persisted.
|
||||
# @TEST_EDGE: key_hash is unique and indexed
|
||||
# @TEST_EDGE: prefix is exactly 11 chars ("ssk_" + 7)
|
||||
# @TEST_EDGE: active defaults to True
|
||||
import pytest
|
||||
from datetime import datetime, timezone
|
||||
|
||||
|
||||
# #region clean_db [C:1] [TYPE Fixture]
|
||||
# @BRIEF In-memory SQLite fixture for APIKey model tests.
|
||||
@pytest.fixture
|
||||
def clean_db():
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from sqlalchemy.pool import StaticPool
|
||||
from src.models.mapping import Base
|
||||
|
||||
engine = create_engine(
|
||||
"sqlite:///:memory:",
|
||||
poolclass=StaticPool,
|
||||
connect_args={"check_same_thread": False},
|
||||
)
|
||||
Base.metadata.create_all(engine)
|
||||
Session = sessionmaker(bind=engine)
|
||||
session = Session()
|
||||
yield session
|
||||
session.close()
|
||||
# #endregion clean_db
|
||||
|
||||
|
||||
class TestAPIKeyModel:
|
||||
"""Verify APIKey model fields and constraints."""
|
||||
|
||||
# #region test_create_api_key [C:2] [TYPE Function]
|
||||
# @BRIEF Create APIKey with required fields and verify defaults.
|
||||
def test_create_api_key(self, clean_db):
|
||||
from src.models.api_key import APIKey
|
||||
|
||||
api_key = APIKey(
|
||||
key_hash="a" * 64,
|
||||
prefix="ssk_abc1234",
|
||||
name="Test Key",
|
||||
permissions=["maintenance:start"],
|
||||
)
|
||||
clean_db.add(api_key)
|
||||
clean_db.commit()
|
||||
clean_db.refresh(api_key)
|
||||
|
||||
assert api_key.id is not None
|
||||
assert api_key.key_hash == "a" * 64
|
||||
assert api_key.prefix == "ssk_abc1234"
|
||||
assert api_key.name == "Test Key"
|
||||
assert api_key.permissions == ["maintenance:start"]
|
||||
assert api_key.active is True
|
||||
assert api_key.created_at is not None
|
||||
assert api_key.environment_id is None
|
||||
assert api_key.expires_at is None
|
||||
assert api_key.last_used_at is None
|
||||
# #endregion test_create_api_key
|
||||
|
||||
# #region test_key_hash_unique [C:2] [TYPE Function]
|
||||
# @BRIEF key_hash is unique — duplicate raises IntegrityError.
|
||||
def test_key_hash_unique(self, clean_db):
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
from src.models.api_key import APIKey
|
||||
|
||||
clean_db.add(APIKey(
|
||||
key_hash="b" * 64,
|
||||
prefix="ssk_xyz7890",
|
||||
name="Key 1",
|
||||
permissions=["maintenance:start"],
|
||||
))
|
||||
clean_db.commit()
|
||||
|
||||
with pytest.raises(IntegrityError):
|
||||
clean_db.add(APIKey(
|
||||
key_hash="b" * 64, # Same hash
|
||||
prefix="ssk_def5678",
|
||||
name="Key 2",
|
||||
permissions=["maintenance:end"],
|
||||
))
|
||||
clean_db.commit()
|
||||
# #endregion test_key_hash_unique
|
||||
|
||||
# #region test_prefix_length [C:2] [TYPE Function]
|
||||
# @BRIEF prefix is exactly 11 characters.
|
||||
def test_prefix_length(self, clean_db):
|
||||
from src.models.api_key import APIKey
|
||||
|
||||
api_key = APIKey(
|
||||
key_hash="c" * 64,
|
||||
prefix="ssk_short1", # 10 chars — should be ok in SQL
|
||||
name="Short Prefix",
|
||||
permissions=["maintenance:start"],
|
||||
)
|
||||
clean_db.add(api_key)
|
||||
clean_db.commit()
|
||||
# The DB allows any prefix length; the app layer enforces 11 chars
|
||||
# via generate_api_key. Model just stores what it's given.
|
||||
assert len(api_key.prefix) <= 11
|
||||
# #endregion test_prefix_length
|
||||
|
||||
# #region test_active_default_true [C:2] [TYPE Function]
|
||||
# @BRIEF active column defaults to True.
|
||||
def test_active_default_true(self, clean_db):
|
||||
from src.models.api_key import APIKey
|
||||
|
||||
api_key = APIKey(
|
||||
key_hash="d" * 64,
|
||||
prefix="ssk_def1234",
|
||||
name="Default Active",
|
||||
permissions=[],
|
||||
)
|
||||
clean_db.add(api_key)
|
||||
clean_db.commit()
|
||||
|
||||
assert api_key.active is True
|
||||
# #endregion test_active_default_true
|
||||
# #endregion TestAPIKeyModel
|
||||
Reference in New Issue
Block a user