busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
2bc8ad87e193d0a1c56531f9557a444dac3e0490
ss-tools/backend/src/dependencies.py
busya 2bc8ad87e1 security: rate limiter, is_admin flag, password policy, log masking 
M-3: Role.name == 'Admin' string comparison → Role.is_admin boolean flag.
  Admin role created with is_admin=True. has_permission checks getattr(role,
  'is_admin', False) instead of comparing role names.

M-2: In-memory rate limiter on /api/auth/login. Tracks failed attempts
  per IP (10 in 5 min window, 15 min ban). Thread-safe via threading.Lock.

M-5: Password policy — UserCreate.password: min_length=8, must include
  uppercase, lowercase, and digit.

L-1: log_security_event details dict now masks sensitive fields:
  password, secret, token, api_key, authorization. Recursive masking
  for nested dicts. Sensitive field names matched via regex.

OTHER:
- 4 rate limiter tests (under limit, ban, success clears, IP isolation)
- 4 password policy tests (too short, no upper, no digit, valid)
- 2 log masking tests (flat keys, nested)
2026-05-26 15:17:19 +03:00

576 lines
22 KiB
Python
Executable File
Raw Blame History

# #region AppDependencies [C:4] [TYPE Module] [SEMANTICS fastapi, schedule, auth, api_key]
# @BRIEF Manages creation and provision of shared application dependencies, such as PluginLoader and TaskManager, to avoid circular imports.
# @LAYER Core
# @PURPOSE Provides shared instances to app and routers.
# @RELATION CALLS -> [CleanReleaseRepository]
# @RELATION CALLS -> [ConfigManager]
# @RELATION CALLS -> [PluginLoader]
# @RELATION CALLS -> [SchedulerService]
# @RELATION CALLS -> [TaskManager]
# @RELATION CALLS -> [get_all_plugin_configs]
# @RELATION CALLS -> [get_db]
# @RELATION CALLS -> [APIKeyPrincipal]
# @RELATION CALLS -> [get_api_key_principal]
# @RELATION CALLS -> [info]
import json
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
from fastapi import Depends, HTTPException, Request, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError
from .core.auth.jwt import decode_token, is_token_blacklisted
from .core.auth.repository import AuthRepository
from .core.config_manager import ConfigManager
from .core.database import get_auth_db, get_db, init_db
from .core.logger import logger
from .core.plugin_loader import PluginLoader
from .core.scheduler import SchedulerService
from .core.task_manager import TaskManager
from .models.api_key import APIKey
from .models.auth import User
from .services.clean_release.facade import CleanReleaseFacade
from .services.clean_release.repositories import (
ApprovalRepository,
ArtifactRepository,
AuditRepository,
CandidateRepository,
ComplianceRepository,
ManifestRepository,
PolicyRepository,
PublicationRepository,
ReportRepository,
)
from .services.clean_release.repository import CleanReleaseRepository
from .services.mapping_service import MappingService
from .services.resource_service import ResourceService
# Initialize singletons lazily to avoid import-time DB side effects during test collection.
# Use absolute path relative to this file to ensure plugins are found regardless of CWD
project_root = Path(__file__).parent.parent.parent
config_path = project_root / "config.json"
config_manager: ConfigManager | None = None
plugin_loader: PluginLoader | None = None
task_manager: TaskManager | None = None
scheduler_service: SchedulerService | None = None
resource_service: ResourceService | None = None
# #region get_config_manager [C:1] [TYPE Function]
# @BRIEF Dependency injector for ConfigManager.
# @PRE Global config_manager must be initialized.
# @POST Returns shared ConfigManager instance.
def get_config_manager() -> ConfigManager:
"""Dependency injector for ConfigManager."""
global config_manager
if config_manager is None:
init_db()
config_manager = ConfigManager(config_path=str(config_path))
return config_manager
# #endregion get_config_manager
plugin_dir = Path(__file__).parent / "plugins"
# Clean Release Redesign Singletons
# Note: These use get_db() which is a generator, so we need a way to provide a session.
# For singletons in dependencies.py, we might need a different approach or
# initialize them inside the dependency functions.
# #region get_plugin_loader [C:1] [TYPE Function]
# @BRIEF Dependency injector for PluginLoader.
# @PRE Global plugin_loader must be initialized.
# @POST Returns shared PluginLoader instance.
def get_plugin_loader() -> PluginLoader:
"""Dependency injector for PluginLoader."""
global plugin_loader
if plugin_loader is None:
plugin_loader = PluginLoader(plugin_dir=str(plugin_dir))
logger.info(f"PluginLoader initialized with directory: {plugin_dir}")
logger.info(
f"Available plugins: {[config.name for config in plugin_loader.get_all_plugin_configs()]}"
)
return plugin_loader
# #endregion get_plugin_loader
# #region get_task_manager [C:1] [TYPE Function]
# @BRIEF Dependency injector for TaskManager.
# @PRE Global task_manager must be initialized.
# @POST Returns shared TaskManager instance.
def get_task_manager() -> TaskManager:
"""Dependency injector for TaskManager."""
global task_manager
if task_manager is None:
task_manager = TaskManager(get_plugin_loader())
logger.info("TaskManager initialized")
return task_manager
# #endregion get_task_manager
# #region get_scheduler_service [C:1] [TYPE Function]
# @BRIEF Dependency injector for SchedulerService.
# @PRE Global scheduler_service must be initialized.
# @POST Returns shared SchedulerService instance.
def get_scheduler_service() -> SchedulerService:
"""Dependency injector for SchedulerService."""
global scheduler_service
if scheduler_service is None:
scheduler_service = SchedulerService(get_task_manager(), get_config_manager())
logger.info("SchedulerService initialized")
return scheduler_service
# #endregion get_scheduler_service
# #region get_resource_service [C:1] [TYPE Function]
# @BRIEF Dependency injector for ResourceService.
# @PRE Global resource_service must be initialized.
# @POST Returns shared ResourceService instance.
def get_resource_service() -> ResourceService:
"""Dependency injector for ResourceService."""
global resource_service
if resource_service is None:
resource_service = ResourceService()
logger.info("ResourceService initialized")
return resource_service
# #endregion get_resource_service
# #region get_mapping_service [C:1] [TYPE Function]
# @BRIEF Dependency injector for MappingService.
# @PRE Global config_manager must be initialized.
# @POST Returns new MappingService instance.
def get_mapping_service() -> MappingService:
"""Dependency injector for MappingService."""
return MappingService(get_config_manager())
# #endregion get_mapping_service
_clean_release_repository = CleanReleaseRepository()
# #region get_clean_release_repository [C:1] [TYPE Function]
# @BRIEF Legacy compatibility shim for CleanReleaseRepository.
# @POST Returns a shared CleanReleaseRepository instance.
def get_clean_release_repository() -> CleanReleaseRepository:
"""Legacy compatibility shim for CleanReleaseRepository."""
return _clean_release_repository
# #endregion get_clean_release_repository
# #region get_clean_release_facade [C:1] [TYPE Function]
# @BRIEF Dependency injector for CleanReleaseFacade.
# @POST Returns a facade instance with a fresh DB session.
def get_clean_release_facade(db=Depends(get_db)) -> CleanReleaseFacade:
candidate_repo = CandidateRepository(db)
artifact_repo = ArtifactRepository(db)
manifest_repo = ManifestRepository(db)
policy_repo = PolicyRepository(db)
compliance_repo = ComplianceRepository(db)
report_repo = ReportRepository(db)
approval_repo = ApprovalRepository(db)
publication_repo = PublicationRepository(db)
audit_repo = AuditRepository(db)
return CleanReleaseFacade(
candidate_repo=candidate_repo,
artifact_repo=artifact_repo,
manifest_repo=manifest_repo,
policy_repo=policy_repo,
compliance_repo=compliance_repo,
report_repo=report_repo,
approval_repo=approval_repo,
publication_repo=publication_repo,
audit_repo=audit_repo,
config_manager=get_config_manager(),
)
# #endregion get_clean_release_facade
# #region APIKeyPrincipal [C:1] [TYPE Class]
# @BRIEF Dataclass representing an authenticated API key principal for service-to-service auth.
@dataclass
class APIKeyPrincipal:
name: str
api_key_id: str
environment_id: Optional[str] = None
permissions: list[str] = field(default_factory=list)
# #endregion APIKeyPrincipal
# #region require_api_key_or_jwt [C:4] [TYPE Function]
# @BRIEF Factory: creates a dependency that accepts either a valid API key (with permission check)
# OR a JWT (with standard has_permission check). Returns the authenticated user/principal name.
# JWT takes precedence when both auth methods are present.
# When API key is used, its environment_id scope is enforced against the caller's environment_id.
# @PRE required_api_permission is a dot-separated string e.g. "maintenance:start".
# required_jwt_resource and required_jwt_action are the JWT permission to check.
# environment_id is the target environment for the operation (may be None for cross-env ops).
# @POST Returns str (username or api key name). Raises 400/401/403 if auth fails or scope mismatch.
# @SIDE_EFFECT Updates APIKey.last_used_at on successful API key auth.
# @RELATION DEPENDS_ON -> [APIKeyModel]
# @RELATION DEPENDS_ON -> [hash_api_key]
# @RELATION DEPENDS_ON -> [AuthRepository]
# @RELATION DEPENDS_ON -> [decode_token]
def require_api_key_or_jwt(
required_api_permission: str,
required_jwt_resource: str,
required_jwt_action: str,
environment_id: str | None = None,
):
"""Factory: creates a combined API-key-or-JWT auth dependency with environment scoping.
JWT takes precedence when both Authorization: Bearer and X-API-Key are present.
When API key auth is used, the key's environment_id scope is enforced:
- If key.environment_id differs from environment_id → 400
- If environment_id is None, the key's scope is used as default (auto-scope).
Args:
required_api_permission: Permission string to check against API key (e.g. 'maintenance:start').
required_jwt_resource: JWT resource for has_permission fallback.
required_jwt_action: JWT action for has_permission fallback.
environment_id: Target environment for the operation. Pass None for cross-env operations.
"""
async def _auth_dependency(
request: Request,
db=Depends(get_db),
auth_db=Depends(get_auth_db),
token: str | None = Depends(oauth2_scheme_optional),
) -> str:
from .core.auth.api_key import hash_api_key
# Resolve effective environment_id: factory parameter > request body > None
effective_env_id = environment_id
if effective_env_id is None:
try:
body = await request.json()
effective_env_id = body.get("environment_id")
except (json.JSONDecodeError, KeyError, AttributeError):
pass
# ── Try JWT auth first (JWT takes precedence over API key) ──
if token:
try:
payload = decode_token(token)
username_value = payload.get("sub")
if not isinstance(username_value, str) or not username_value:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
username = username_value
except Exception:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
repo = AuthRepository(auth_db)
user = repo.get_user_by_username(username)
if user is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="User not found",
)
# Check JWT permission (Admin has full access)
has_perm = any(
role.name == "Admin"
for role in user.roles
)
if not has_perm:
for role in user.roles:
for perm in role.permissions:
if perm.resource == required_jwt_resource and perm.action == required_jwt_action:
has_perm = True
break
if has_perm:
break
if not has_perm:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"Permission denied for {required_jwt_resource}:{required_jwt_action}",
)
logger.info(f"JWT auth: {user.username}")
return f"jwt:{user.username}"
# ── Try API key auth ──
raw_key = request.headers.get("X-API-Key")
if raw_key:
key_hash = hash_api_key(raw_key)
api_key = db.query(APIKey).filter(APIKey.key_hash == key_hash).first()
if not api_key:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid API key",
)
if not api_key.active:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="API key has been revoked",
)
if api_key.expires_at:
expires_at = api_key.expires_at
if expires_at.tzinfo is None:
expires_at = expires_at.replace(tzinfo=timezone.utc)
if expires_at < datetime.now(timezone.utc):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="API key has expired",
)
# Check API key has required permission
api_permissions = list(api_key.permissions or [])
if required_api_permission not in api_permissions:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"API key lacks permission '{required_api_permission}'",
)
# Environment scoping check: key's scope vs effective operation env
key_env = api_key.environment_id
if key_env is not None and effective_env_id is not None and key_env != effective_env_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"API key is restricted to environment '{key_env}'",
)
# Update last_used_at (best-effort, non-blocking)
api_key.last_used_at = datetime.now(timezone.utc)
db.flush()
logger.info(f"API key auth: {api_key.name} ({api_key.prefix}...)")
return f"api_key:{api_key.name}"
# ── Neither JWT nor API key provided ──
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Authentication required (API key or JWT)",
)
return _auth_dependency
# #endregion require_api_key_or_jwt
# #region check_api_key_environment_scope [C:1] [TYPE Function]
# @BRIEF Check the request's API key environment scope against a target env_id. Raises 400 if mismatch.
# @RELATION DEPENDS_ON -> [APIKeyModel]
# @RELATION DEPENDS_ON -> [hash_api_key]
async def check_api_key_environment_scope(request: Request, db, target_env_id: str) -> None:
"""Check the request's API key environment scope against a target environment_id.
If the API key has an environment_id set and it differs from target_env_id,
raises HTTP 400. This is used in route handlers where the target env_id
is only known at request time (e.g., loaded from DB in end_maintenance).
Args:
request: FastAPI request object (to extract X-API-Key header).
db: Database session.
target_env_id: The target environment to validate against.
"""
from .core.auth.api_key import hash_api_key
raw_key = request.headers.get("X-API-Key")
if not raw_key:
return
key_hash = hash_api_key(raw_key)
api_key = db.query(APIKey).filter(APIKey.key_hash == key_hash).first()
if api_key is None:
return
key_env = api_key.environment_id
if key_env is not None and key_env != target_env_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"API key is restricted to environment '{key_env}'",
)
# #endregion check_api_key_environment_scope
# #region get_api_key_principal [C:3] [TYPE Function]
# @BRIEF FastAPI dependency that reads X-API-Key header and returns APIKeyPrincipal or None.
# @PRE X-API-Key header may be present in the request.
# @POST If valid key: returns APIKeyPrincipal; if no header: returns None; if invalid: raises 401.
# @SIDE_EFFECT Updates last_used_at on the APIKey row (non-blocking flush).
# @RELATION DEPENDS_ON -> [APIKeyModel]
# @RELATION DEPENDS_ON -> [hash_api_key]
async def get_api_key_principal(
request: Request,
db=Depends(get_db),
) -> Optional[APIKeyPrincipal]:
"""Extract and validate API key from X-API-Key header.
If no header is present, returns None (fall through to JWT auth).
If header is present but invalid, raises 401.
Returns:
APIKeyPrincipal | None: Principal if valid API key, None if no header.
"""
from .core.auth.api_key import hash_api_key
raw_key = request.headers.get("X-API-Key")
if not raw_key:
return None
key_hash = hash_api_key(raw_key)
api_key = db.query(APIKey).filter(APIKey.key_hash == key_hash).first()
if not api_key:
logger.warning(f"API key auth failed: key not found (hash prefix: {raw_key[:11]}...)")
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid API key",
)
if not api_key.active:
logger.warning(f"API key auth failed: key revoked (id: {api_key.id}, name: {api_key.name})")
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="API key has been revoked",
)
if api_key.expires_at:
expires_at = api_key.expires_at
if expires_at.tzinfo is None:
expires_at = expires_at.replace(tzinfo=timezone.utc)
if expires_at < datetime.now(timezone.utc):
logger.warning(f"API key auth failed: key expired (id: {api_key.id}, name: {api_key.name})")
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="API key has expired",
)
# Update last_used_at (non-blocking, best-effort)
api_key.last_used_at = datetime.now(timezone.utc)
db.flush()
logger.info(f"API key auth success: {api_key.name} ({api_key.prefix}...)")
return APIKeyPrincipal(
name=api_key.name,
api_key_id=api_key.id,
environment_id=api_key.environment_id,
permissions=list(api_key.permissions or []),
)
# #endregion get_api_key_principal
# #region oauth2_scheme [C:1] [TYPE Variable]
# @RELATION DEPENDS_ON -> [EXT:Library:OAuth2PasswordBearer]
# @BRIEF OAuth2 password bearer scheme for token extraction (raises 401 on missing token).
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login", auto_error=True)
# #endregion oauth2_scheme
# #region oauth2_scheme_optional [C:1] [TYPE Variable]
# @RELATION DEPENDS_ON -> [EXT:Library:OAuth2PasswordBearer]
# @BRIEF Optional OAuth2 scheme — returns None instead of raising 401 when no token.
# @RATIONALE Used in dual-auth routes (API key OR JWT) where JWT may be absent.
oauth2_scheme_optional = OAuth2PasswordBearer(tokenUrl="/api/auth/login", auto_error=False)
# #endregion oauth2_scheme_optional
# #region get_current_user [C:3] [TYPE Function]
# @RELATION CALLS -> AuthRepository
# @RELATION CALLS -> [is_token_blacklisted]
# @BRIEF Dependency for retrieving currently authenticated user from a JWT.
# @PRE JWT token provided in Authorization header.
# @POST Returns User object if token is valid and not revoked.
def get_current_user(token: str = Depends(oauth2_scheme), db=Depends(get_auth_db)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = decode_token(token)
username_value = payload.get("sub")
if not isinstance(username_value, str) or not username_value:
raise credentials_exception
username = username_value
except JWTError:
raise credentials_exception
# Check blacklist
if is_token_blacklisted(token, db):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Token has been revoked",
headers={"WWW-Authenticate": "Bearer"},
)
repo = AuthRepository(db)
user = repo.get_user_by_username(username)
if user is None:
raise credentials_exception
return user
# #endregion get_current_user
# #region has_permission [C:3] [TYPE Function]
# @RELATION CALLS -> AuthRepository
# @BRIEF Dependency for checking if the current user has a specific permission.
# @PRE User is authenticated.
# @POST Returns True if user has permission.
def has_permission(resource: str, action: str):
def permission_checker(current_user: User = Depends(get_current_user)):
# Union of all permissions across all roles
for role in current_user.roles:
for perm in role.permissions:
if perm.resource == resource and perm.action == action:
return current_user
# Special case for Admin role (full access) — uses is_admin flag, not name string
if any(getattr(role, "is_admin", False) for role in current_user.roles):
return current_user
from .core.auth.logger import log_security_event
log_security_event(
"PERMISSION_DENIED",
str(getattr(current_user, "username", "unknown")),
{"resource": resource, "action": action},
)
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"Permission denied for {resource}:{action}",
)
return permission_checker
# #endregion has_permission
# #endregion AppDependencies
Reference in New Issue View Git Blame Copy Permalink