Lars
30dc30c7aa
Some checks failed
Deploy Development / deploy (push) Successful in 43s
Test Suite / pytest-backend (push) Failing after 0s
Test Suite / lint-backend (push) Successful in 0s
Test Suite / build-frontend (push) Successful in 14s
Test Suite / k6 /health Baseline (push) Failing after 4m0s
Test Suite / playwright-tests (push) Failing after 3m41s
- Introduced `email_verified` and `account_state` attributes in the `TenantContext` to improve user state management. - Updated the `resolve_tenant_context` function to dynamically fetch `email_verified` status from the database and determine `account_state` based on user roles and memberships. - Implemented `assert_min_account_state` checks across various endpoints to enforce access control based on user account status. - Incremented version to 1.1.0 in version.py to reflect these enhancements in tenant context management and access control.
65 lines
2.0 KiB
Python
65 lines
2.0 KiB
Python
"""
|
|
JSON-Log für Capability-Checks (M3 Phase 2 — analog club_feature_logger).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Optional
|
|
|
|
|
|
def _log_dir() -> Path:
|
|
custom = (os.getenv("CAPABILITY_LOG_DIR") or "").strip()
|
|
if custom:
|
|
return Path(custom)
|
|
return Path("/app/logs")
|
|
|
|
|
|
capability_logger = logging.getLogger("shinkan.capability_usage")
|
|
capability_logger.setLevel(logging.INFO)
|
|
capability_logger.propagate = False
|
|
|
|
if not capability_logger.handlers:
|
|
log_dir = _log_dir()
|
|
try:
|
|
log_dir.mkdir(parents=True, exist_ok=True)
|
|
log_file = log_dir / "capability-usage.log"
|
|
file_handler = logging.FileHandler(log_file, encoding="utf-8")
|
|
file_handler.setLevel(logging.INFO)
|
|
file_handler.setFormatter(logging.Formatter("%(message)s"))
|
|
capability_logger.addHandler(file_handler)
|
|
except OSError:
|
|
stream_handler = logging.StreamHandler()
|
|
stream_handler.setFormatter(logging.Formatter("[capability-usage] %(message)s"))
|
|
capability_logger.addHandler(stream_handler)
|
|
|
|
|
|
def log_capability_check(
|
|
*,
|
|
club_id: Optional[int],
|
|
profile_id: Optional[int],
|
|
capability_id: str,
|
|
action: str,
|
|
result: Dict[str, Any],
|
|
endpoint: Optional[str] = None,
|
|
phase: str = "probe",
|
|
) -> None:
|
|
entry = {
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
"club_id": club_id,
|
|
"profile_id": profile_id,
|
|
"capability": capability_id,
|
|
"action": action,
|
|
"endpoint": endpoint,
|
|
"phase": phase,
|
|
"allowed": result.get("allowed", True),
|
|
"reason": result.get("reason", "unknown"),
|
|
"account_state": result.get("account_state"),
|
|
"club_roles": result.get("club_roles"),
|
|
"enforcement": os.getenv("CAPABILITY_ENFORCE", "0") == "1",
|
|
}
|
|
capability_logger.info(json.dumps(entry, ensure_ascii=False))
|