Implement Club Feature Quota Bypass and Update Versioning
Some checks failed
Deploy Development / deploy (push) Successful in 43s
Test Suite / pytest-backend (push) Failing after 2s
Test Suite / lint-backend (push) Successful in 0s
Test Suite / build-frontend (push) Successful in 13s
Test Suite / k6 /health Baseline (push) Successful in 42s
Test Suite / playwright-tests (push) Successful in 1m19s

- Added support for club feature quota bypass based on portal roles and profile grants in the capabilities check.
- Introduced new functions to handle quota bypass logic in club feature access and consumption.
- Updated the FeatureUsageBadge component to reflect platform exemptions for features.
- Incremented application version to 0.8.195 and database schema version to 20260606083 to reflect these changes.
- Enhanced backend routers to include new logic for consuming club features during AI-related actions.
This commit is contained in:
Lars 2026-06-07 07:43:35 +02:00
parent fa10450315
commit 8404a42b6c
16 changed files with 1063 additions and 41 deletions

View File

@ -81,6 +81,49 @@ def check_capability(
domain = (cap.get("domain") or "").strip().lower()
# Kontingent-Bypass (konfigurierbar per portal_role / profile grants, ohne Plattform-Admin-Pflicht)
if domain == "quota_bypass":
role_lc = (tenant.global_role or "").lower()
cur.execute(
"""
SELECT 1 FROM portal_role_capability_grants
WHERE portal_role = %s AND capability_id = %s
LIMIT 1
""",
(role_lc, capability_id),
)
if cur.fetchone():
return {
"allowed": True,
"reason": "quota_bypass_portal_grant",
"account_state": account_state,
"club_roles": club_roles,
"linked_feature_id": cap.get("linked_feature_id"),
}
cur.execute(
"""
SELECT 1 FROM profile_capability_grants
WHERE profile_id = %s AND capability_id = %s
LIMIT 1
""",
(tenant.profile_id, capability_id),
)
if cur.fetchone():
return {
"allowed": True,
"reason": "quota_bypass_profile_grant",
"account_state": account_state,
"club_roles": club_roles,
"linked_feature_id": cap.get("linked_feature_id"),
}
return {
"allowed": False,
"reason": "quota_bypass_denied",
"account_state": account_state,
"club_roles": club_roles,
"linked_feature_id": cap.get("linked_feature_id"),
}
# Plattform-Capabilities
if domain == "platform" or capability_id.startswith("platform."):
role_lc = (tenant.global_role or "").lower()
@ -100,6 +143,15 @@ def check_capability(
""",
(role_lc, capability_id),
)
if not cur.fetchone():
cur.execute(
"""
SELECT 1 FROM profile_capability_grants
WHERE profile_id = %s AND capability_id = %s
LIMIT 1
""",
(tenant.profile_id, capability_id),
)
if not cur.fetchone():
return {
"allowed": False,

View File

@ -328,6 +328,7 @@ def probe_club_feature_access(
action: str,
club_id: Optional[int] = None,
profile_id: Optional[int] = None,
portal_role: Optional[str] = None,
endpoint: Optional[str] = None,
conn=None,
) -> Dict[str, Any]:
@ -358,11 +359,29 @@ def probe_club_feature_access(
)
return access
def _resolve_access(connection):
from club_quota_bypass import is_club_feature_quota_bypassed, quota_bypass_access
cur = get_cursor(connection)
if is_club_feature_quota_bypassed(
cur,
profile_id=profile_id,
portal_role=portal_role,
feature_id=feature_id,
):
plan_id = get_effective_club_plan(cur, int(club_id))
return quota_bypass_access(
feature_id=feature_id,
club_id=int(club_id),
plan_id=plan_id,
)
return check_club_feature_access(club_id, feature_id, conn=connection)
if conn is not None:
access = check_club_feature_access(club_id, feature_id, conn=conn)
access = _resolve_access(conn)
else:
with get_db() as c:
access = check_club_feature_access(club_id, feature_id, conn=c)
access = _resolve_access(c)
log_club_feature_usage(
club_id=club_id,
@ -387,6 +406,58 @@ def probe_club_feature_access(
return access
def consume_club_feature(
*,
feature_id: str,
club_id: Optional[int],
profile_id: Optional[int] = None,
portal_role: Optional[str] = None,
action: Optional[str] = None,
amount: int = 1,
conn=None,
) -> None:
"""
Phase 4 (M5): Zähler nach erfolgreichem Verbrauch erhöhen.
Nur wenn club_id gesetzt (Vereins-Kontingent); amount = Anzahl LLM/API-Verbrauchseinheiten.
Plattform-Ausnahmen (superadmin, konfigurierte Rollen/Profile) werden nicht gezählt.
"""
if club_id is None:
return
def _is_exempt(connection) -> bool:
from club_quota_bypass import is_club_feature_quota_bypassed
cur = get_cursor(connection)
return is_club_feature_quota_bypassed(
cur,
profile_id=profile_id,
portal_role=portal_role,
feature_id=feature_id,
)
if conn is not None:
if _is_exempt(conn):
return
else:
with get_db() as c:
if _is_exempt(c):
return
try:
n = int(amount)
except (TypeError, ValueError):
n = 1
if n < 1:
return
for _ in range(n):
increment_club_feature_usage(
int(club_id),
feature_id,
profile_id=profile_id,
action=action,
conn=conn,
)
def increment_club_feature_usage(
club_id: int,
feature_id: str,

View File

@ -0,0 +1,180 @@
"""
Vereins-Kontingent-Bypass über das Capability-System (kein Parallel-Rechtemodell).
Capabilities:
- platform.club_quota.bypass alle Vereins-Features (Portal-Admin, Grant via portal_role)
- platform.club_quota.bypass.{feature_id} ein Feature (domain quota_bypass, auch für Nicht-Admins per Grant)
"""
from __future__ import annotations
from typing import Any, Dict, List, Optional, TYPE_CHECKING
if TYPE_CHECKING:
from tenant_context import TenantContext
QUOTA_BYPASS_ALL = "platform.club_quota.bypass"
QUOTA_BYPASS_FEATURE_PREFIX = "platform.club_quota.bypass."
def quota_bypass_capability_id_for_feature(feature_id: str) -> str:
return f"{QUOTA_BYPASS_FEATURE_PREFIX}{feature_id}"
def ensure_quota_bypass_capability(cur, feature_id: str) -> str:
"""Legt feature-spezifische Bypass-Capability an falls nötig."""
cap_id = quota_bypass_capability_id_for_feature(feature_id)
cur.execute(
"""
INSERT INTO capabilities (id, name, domain, min_account_state, linked_feature_id)
VALUES (%s, %s, 'quota_bypass', 'active_member', %s)
ON CONFLICT (id) DO NOTHING
""",
(cap_id, f"Vereins-Kontingent umgehen: {feature_id}", feature_id),
)
return cap_id
def _bypass_capability_ids(cur, feature_id: str) -> List[str]:
ids: List[str] = [QUOTA_BYPASS_ALL, quota_bypass_capability_id_for_feature(feature_id)]
cur.execute(
"""
SELECT id FROM capabilities
WHERE active = true
AND domain = 'quota_bypass'
AND linked_feature_id = %s
AND id <> %s
""",
(feature_id, quota_bypass_capability_id_for_feature(feature_id)),
)
for row in cur.fetchall():
cid = row.get("id")
if cid and cid not in ids:
ids.append(str(cid))
return ids
def _portal_role_has_grant(cur, portal_role: str, capability_id: str) -> bool:
role = (portal_role or "").strip().lower()
if not role:
return False
cur.execute(
"""
SELECT 1 FROM portal_role_capability_grants
WHERE portal_role = %s AND capability_id = %s
LIMIT 1
""",
(role, capability_id),
)
return cur.fetchone() is not None
def _profile_has_grant(cur, profile_id: int, capability_id: str) -> bool:
cur.execute(
"""
SELECT 1 FROM profile_capability_grants
WHERE profile_id = %s AND capability_id = %s
LIMIT 1
""",
(int(profile_id), capability_id),
)
return cur.fetchone() is not None
def is_club_feature_quota_bypassed(
cur,
*,
profile_id: Optional[int],
portal_role: Optional[str],
feature_id: str,
tenant: Optional["TenantContext"] = None,
) -> bool:
"""
True wenn ein konfigurierter Capability-Grant das Vereins-Kontingent für feature_id umgeht.
"""
if tenant is not None:
from capabilities import check_capability
for cap_id in _bypass_capability_ids(cur, feature_id):
if check_capability(cur, tenant, cap_id).get("allowed"):
return True
return False
for cap_id in _bypass_capability_ids(cur, feature_id):
if _portal_role_has_grant(cur, portal_role or "", cap_id):
return True
if profile_id is not None and _profile_has_grant(cur, int(profile_id), cap_id):
return True
return False
def quota_bypass_access(
*,
feature_id: str,
club_id: Optional[int] = None,
plan_id: Optional[str] = None,
capability_id: Optional[str] = None,
) -> Dict[str, Any]:
return {
"allowed": True,
"limit": None,
"used": 0,
"remaining": None,
"reason": "capability_quota_bypass",
"platform_exempt": True,
"quota_bypass_capability": capability_id,
"plan_id": plan_id,
"club_id": club_id,
"feature_id": feature_id,
}
def list_quota_bypass_grants(cur) -> Dict[str, Any]:
"""Admin: alle Grants zu Kontingent-Bypass-Capabilities."""
cur.execute(
"""
SELECT g.portal_role, g.capability_id, c.name AS capability_name,
c.linked_feature_id, c.domain
FROM portal_role_capability_grants g
INNER JOIN capabilities c ON c.id = g.capability_id
WHERE g.capability_id = %s
OR g.capability_id LIKE %s
OR c.domain = 'quota_bypass'
ORDER BY g.portal_role, g.capability_id
""",
(QUOTA_BYPASS_ALL, f"{QUOTA_BYPASS_FEATURE_PREFIX}%"),
)
portal_grants = [dict(r) for r in cur.fetchall()]
cur.execute(
"""
SELECT g.profile_id, p.email, p.name AS profile_name,
g.capability_id, c.name AS capability_name, c.linked_feature_id,
g.reason, g.granted_by_profile_id, g.created_at
FROM profile_capability_grants g
INNER JOIN profiles p ON p.id = g.profile_id
INNER JOIN capabilities c ON c.id = g.capability_id
WHERE g.capability_id = %s
OR g.capability_id LIKE %s
OR c.domain = 'quota_bypass'
ORDER BY g.profile_id, g.capability_id
""",
(QUOTA_BYPASS_ALL, f"{QUOTA_BYPASS_FEATURE_PREFIX}%"),
)
profile_grants = [dict(r) for r in cur.fetchall()]
cur.execute(
"""
SELECT id, name, domain, linked_feature_id
FROM capabilities
WHERE id = %s OR id LIKE %s OR domain = 'quota_bypass'
ORDER BY id
""",
(QUOTA_BYPASS_ALL, f"{QUOTA_BYPASS_FEATURE_PREFIX}%"),
)
capabilities = [dict(r) for r in cur.fetchall()]
return {
"capabilities": capabilities,
"portal_role_grants": portal_grants,
"profile_grants": profile_grants,
}

View File

@ -11,6 +11,7 @@ from typing import Any, Dict, Optional, TYPE_CHECKING
from fastapi import HTTPException
from capabilities import club_roles_in_club, resolve_capabilities_map
from club_quota_bypass import is_club_feature_quota_bypassed, quota_bypass_access
from club_features import club_features_map
from club_tenancy import is_platform_admin
from tenant_context import _club_exists
@ -69,6 +70,28 @@ def build_me_entitlements(
raw = club_features_map(cur, target_club)
plan_id = raw.get("plan_id")
for fid, row in (raw.get("features") or {}).items():
if is_club_feature_quota_bypassed(
cur,
profile_id=tenant.profile_id,
portal_role=tenant.global_role,
feature_id=fid,
tenant=tenant,
):
ex = quota_bypass_access(
feature_id=fid,
club_id=target_club,
plan_id=plan_id,
)
features[fid] = {
"allowed": True,
"used": row.get("used"),
"limit": None,
"remaining": None,
"reset_at": _serialize_reset_at(row.get("reset_at")),
"reason": ex.get("reason"),
"platform_exempt": True,
}
else:
features[fid] = {
"allowed": row.get("allowed"),
"used": row.get("used"),
@ -76,6 +99,7 @@ def build_me_entitlements(
"remaining": row.get("remaining"),
"reset_at": _serialize_reset_at(row.get("reset_at")),
"reason": row.get("reason"),
"platform_exempt": False,
}
return {

View File

@ -221,7 +221,7 @@ def read_root():
return out
# Register routers
from routers import auth, profiles, exercises, exercise_progression_graphs, clubs, club_memberships, club_join_requests, club_creation_requests, admin_users, admin_user_content, me_entitlements, platform_media_storage, media_assets, skills, skill_profiles, training_planning, planning_exercise_suggest, dashboard, training_modules, training_framework_programs, catalogs, maturity_models, matrix_stack_bundle, matrix_editor, import_wiki, import_wiki_admin, legal_documents, content_reports, ai_prompts_admin, ai_skill_retrieval_admin, exercise_enrichment_admin
from routers import auth, profiles, exercises, exercise_progression_graphs, clubs, club_memberships, club_join_requests, club_creation_requests, admin_users, admin_user_content, admin_club_feature_exemptions, me_entitlements, platform_media_storage, media_assets, skills, skill_profiles, training_planning, planning_exercise_suggest, dashboard, training_modules, training_framework_programs, catalogs, maturity_models, matrix_stack_bundle, matrix_editor, import_wiki, import_wiki_admin, legal_documents, content_reports, ai_prompts_admin, ai_skill_retrieval_admin, exercise_enrichment_admin
app.include_router(auth.router)
app.include_router(profiles.router)
@ -233,6 +233,7 @@ app.include_router(club_join_requests.router)
app.include_router(club_creation_requests.router)
app.include_router(admin_users.router)
app.include_router(admin_user_content.router)
app.include_router(admin_club_feature_exemptions.router)
app.include_router(me_entitlements.router)
app.include_router(platform_media_storage.router)
app.include_router(media_assets.router)

View File

@ -0,0 +1,36 @@
-- Migration 082: Plattform-/Profil-Ausnahmen vom Vereins-Kontingent (M5+)
-- Superadmin & konfigurierbare Rollen/Profile verbrauchen kein club_feature_usage.
CREATE TABLE IF NOT EXISTS platform_role_club_feature_exemptions (
id SERIAL PRIMARY KEY,
portal_role TEXT NOT NULL,
feature_id TEXT REFERENCES features(id) ON DELETE CASCADE,
note TEXT,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE UNIQUE INDEX IF NOT EXISTS uq_platform_role_club_feat_exempt
ON platform_role_club_feature_exemptions (portal_role, COALESCE(feature_id, '*'));
CREATE TABLE IF NOT EXISTS profile_club_feature_exemptions (
id SERIAL PRIMARY KEY,
profile_id INT NOT NULL REFERENCES profiles(id) ON DELETE CASCADE,
feature_id TEXT REFERENCES features(id) ON DELETE CASCADE,
reason TEXT,
set_by_profile_id INT REFERENCES profiles(id) ON DELETE SET NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE UNIQUE INDEX IF NOT EXISTS uq_profile_club_feat_exempt
ON profile_club_feature_exemptions (profile_id, COALESCE(feature_id, '*'));
CREATE INDEX IF NOT EXISTS idx_profile_club_feat_exempt_profile
ON profile_club_feature_exemptions (profile_id);
-- Superadmin: alle Vereins-Features ohne Kontingent-Verbrauch
INSERT INTO platform_role_club_feature_exemptions (portal_role, feature_id, note)
SELECT 'superadmin', NULL, 'Plattform-Administrator: kein Vereins-Kontingent'
WHERE NOT EXISTS (
SELECT 1 FROM platform_role_club_feature_exemptions
WHERE portal_role = 'superadmin' AND feature_id IS NULL
);

View File

@ -0,0 +1,103 @@
-- Migration 083: Vereins-Kontingent-Bypass über Capability-System (kein Parallel-Schema)
-- Ersetzt platform_role_club_feature_exemptions / profile_club_feature_exemptions aus 082.
-- Einzelprofil-Grants (ergänzt portal_role_capability_grants)
CREATE TABLE IF NOT EXISTS profile_capability_grants (
profile_id INT NOT NULL REFERENCES profiles(id) ON DELETE CASCADE,
capability_id TEXT NOT NULL REFERENCES capabilities(id) ON DELETE CASCADE,
reason TEXT,
granted_by_profile_id INT REFERENCES profiles(id) ON DELETE SET NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
PRIMARY KEY (profile_id, capability_id)
);
CREATE INDEX IF NOT EXISTS idx_profile_capability_grants_cap
ON profile_capability_grants(capability_id);
-- Bypass-Capabilities (CAPABILITY_CATALOG — konfigurierbar via portal/profile grants)
INSERT INTO capabilities (id, name, domain, min_account_state, linked_feature_id)
VALUES
(
'platform.club_quota.bypass',
'Vereins-Kontingent umgehen (alle Features)',
'platform',
'platform_admin',
NULL
)
ON CONFLICT (id) DO NOTHING;
-- Superadmin: alle Plattform-Capabilities inkl. bypass (079-Seed deckt domain=platform ab)
INSERT INTO portal_role_capability_grants (portal_role, capability_id)
SELECT 'superadmin', 'platform.club_quota.bypass'
WHERE NOT EXISTS (
SELECT 1 FROM portal_role_capability_grants
WHERE portal_role = 'superadmin' AND capability_id = 'platform.club_quota.bypass'
);
-- ── Daten aus 082 übernehmen (falls vorhanden) ─────────────────────────────
DO $migrate082$
DECLARE
r RECORD;
cap_id TEXT;
BEGIN
IF NOT EXISTS (
SELECT 1 FROM information_schema.tables
WHERE table_schema = 'public' AND table_name = 'platform_role_club_feature_exemptions'
) THEN
RETURN;
END IF;
FOR r IN
SELECT portal_role, feature_id, note
FROM platform_role_club_feature_exemptions
LOOP
IF r.feature_id IS NULL THEN
cap_id := 'platform.club_quota.bypass';
ELSE
cap_id := 'platform.club_quota.bypass.' || r.feature_id;
INSERT INTO capabilities (id, name, domain, min_account_state, linked_feature_id)
VALUES (
cap_id,
'Vereins-Kontingent umgehen: ' || r.feature_id,
'quota_bypass',
'active_member',
r.feature_id
)
ON CONFLICT (id) DO NOTHING;
END IF;
INSERT INTO portal_role_capability_grants (portal_role, capability_id)
VALUES (lower(trim(r.portal_role)), cap_id)
ON CONFLICT DO NOTHING;
END LOOP;
FOR r IN
SELECT profile_id, feature_id, reason, set_by_profile_id
FROM profile_club_feature_exemptions
LOOP
IF r.feature_id IS NULL THEN
cap_id := 'platform.club_quota.bypass';
ELSE
cap_id := 'platform.club_quota.bypass.' || r.feature_id;
INSERT INTO capabilities (id, name, domain, min_account_state, linked_feature_id)
VALUES (
cap_id,
'Vereins-Kontingent umgehen: ' || r.feature_id,
'quota_bypass',
'active_member',
r.feature_id
)
ON CONFLICT (id) DO NOTHING;
END IF;
INSERT INTO profile_capability_grants (
profile_id, capability_id, reason, granted_by_profile_id
)
VALUES (r.profile_id, cap_id, r.reason, r.set_by_profile_id)
ON CONFLICT DO NOTHING;
END LOOP;
DROP TABLE IF EXISTS profile_club_feature_exemptions;
DROP TABLE IF EXISTS platform_role_club_feature_exemptions;
END
$migrate082$;

View File

@ -0,0 +1,227 @@
"""
Superadmin: Kontingent-Bypass über Capability-Grants (portal_role / profile).
Kein separates Exemption-Schema nutzt portal_role_capability_grants und
profile_capability_grants mit IDs platform.club_quota.bypass[.feature_id].
"""
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, Field
from auth import require_auth
from club_quota_bypass import (
QUOTA_BYPASS_ALL,
ensure_quota_bypass_capability,
list_quota_bypass_grants,
quota_bypass_capability_id_for_feature,
)
from club_tenancy import is_superadmin
from db import get_db, get_cursor, r2d
router = APIRouter(prefix="/api/admin", tags=["admin_capability_grants"])
def _require_superadmin(session: dict) -> None:
if not is_superadmin(session.get("role")):
raise HTTPException(status_code=403, detail="Nur Super-Administratoren")
def _resolve_capability_id(cur, feature_id: Optional[str]) -> str:
fid = (feature_id or "").strip() or None
if not fid:
return QUOTA_BYPASS_ALL
cur.execute("SELECT 1 FROM features WHERE id = %s", (fid,))
if not cur.fetchone():
raise HTTPException(status_code=400, detail="Unbekanntes Feature")
return ensure_quota_bypass_capability(cur, fid)
class PortalGrantBody(BaseModel):
portal_role: str = Field(..., min_length=1, max_length=50)
feature_id: Optional[str] = Field(
None,
description="Feature-ID oder leer = alle Vereins-Features (platform.club_quota.bypass)",
)
class ProfileGrantBody(BaseModel):
feature_id: Optional[str] = Field(None, description="Feature-ID oder leer = alle Features")
reason: Optional[str] = Field(None, max_length=500)
@router.get("/club-feature-exemptions")
def list_club_feature_exemptions(session: dict = Depends(require_auth)):
"""Übersicht Kontingent-Bypass-Grants (Capability-System)."""
_require_superadmin(session)
with get_db() as conn:
cur = get_cursor(conn)
return list_quota_bypass_grants(cur)
@router.get("/capability-grants/club-quota-bypass")
def list_quota_bypass_capability_grants(session: dict = Depends(require_auth)):
"""Alias — gleiche Daten wie /club-feature-exemptions."""
return list_club_feature_exemptions(session)
@router.post("/club-feature-exemptions/roles", status_code=201)
@router.post("/capability-grants/club-quota-bypass/portal-roles", status_code=201)
def add_portal_quota_bypass_grant(body: PortalGrantBody, session: dict = Depends(require_auth)):
_require_superadmin(session)
role = body.portal_role.strip().lower()
with get_db() as conn:
cur = get_cursor(conn)
cap_id = _resolve_capability_id(cur, body.feature_id)
cur.execute(
"""
SELECT 1 FROM portal_role_capability_grants
WHERE portal_role = %s AND capability_id = %s
LIMIT 1
""",
(role, cap_id),
)
if cur.fetchone():
raise HTTPException(status_code=409, detail="Grant existiert bereits")
cur.execute(
"""
INSERT INTO portal_role_capability_grants (portal_role, capability_id)
VALUES (%s, %s)
RETURNING portal_role, capability_id
""",
(role, cap_id),
)
row = cur.fetchone()
conn.commit()
out = r2d(row)
out["capability_id"] = cap_id
if body.feature_id:
out["feature_id"] = body.feature_id.strip()
else:
out["feature_id"] = None
return out
@router.delete("/club-feature-exemptions/roles/{exemption_id}")
def delete_legacy_role_exemption(exemption_id: int, session: dict = Depends(require_auth)):
"""Legacy-Pfad: exemption_id = portal_role_capability_grants nicht unterstützt — 410."""
_require_superadmin(session)
raise HTTPException(
status_code=410,
detail="Bitte DELETE /api/admin/capability-grants/club-quota-bypass/portal-roles nutzen",
)
@router.delete("/capability-grants/club-quota-bypass/portal-roles")
def delete_portal_quota_bypass_grant(
portal_role: str,
capability_id: Optional[str] = None,
feature_id: Optional[str] = None,
session: dict = Depends(require_auth),
):
_require_superadmin(session)
role = portal_role.strip().lower()
cap_id = capability_id
if not cap_id:
cap_id = QUOTA_BYPASS_ALL if not (feature_id or "").strip() else quota_bypass_capability_id_for_feature(
feature_id.strip()
)
with get_db() as conn:
cur = get_cursor(conn)
cur.execute(
"""
DELETE FROM portal_role_capability_grants
WHERE portal_role = %s AND capability_id = %s
RETURNING portal_role, capability_id
""",
(role, cap_id),
)
if not cur.fetchone():
raise HTTPException(status_code=404, detail="Grant nicht gefunden")
conn.commit()
return {"ok": True}
@router.post("/club-feature-exemptions/profiles/{profile_id}", status_code=201)
@router.post("/capability-grants/club-quota-bypass/profiles/{profile_id}", status_code=201)
def add_profile_quota_bypass_grant(
profile_id: int,
body: ProfileGrantBody,
session: dict = Depends(require_auth),
):
_require_superadmin(session)
admin_pid = int(session["profile_id"])
with get_db() as conn:
cur = get_cursor(conn)
cur.execute("SELECT 1 FROM profiles WHERE id = %s", (profile_id,))
if not cur.fetchone():
raise HTTPException(status_code=404, detail="Profil nicht gefunden")
cap_id = _resolve_capability_id(cur, body.feature_id)
cur.execute(
"""
SELECT 1 FROM profile_capability_grants
WHERE profile_id = %s AND capability_id = %s
LIMIT 1
""",
(profile_id, cap_id),
)
if cur.fetchone():
raise HTTPException(status_code=409, detail="Grant existiert bereits")
cur.execute(
"""
INSERT INTO profile_capability_grants (
profile_id, capability_id, reason, granted_by_profile_id
)
VALUES (%s, %s, %s, %s)
RETURNING profile_id, capability_id, reason, granted_by_profile_id, created_at
""",
(profile_id, cap_id, (body.reason or "").strip() or None, admin_pid),
)
row = cur.fetchone()
conn.commit()
return r2d(row)
@router.delete("/club-feature-exemptions/profiles/{exemption_id}")
def delete_legacy_profile_exemption(exemption_id: int, session: dict = Depends(require_auth)):
_require_superadmin(session)
raise HTTPException(
status_code=410,
detail="Bitte DELETE /api/admin/capability-grants/club-quota-bypass/profiles nutzen",
)
@router.delete("/capability-grants/club-quota-bypass/profiles")
def delete_profile_quota_bypass_grant(
profile_id: int,
capability_id: Optional[str] = None,
feature_id: Optional[str] = None,
session: dict = Depends(require_auth),
):
_require_superadmin(session)
cap_id = capability_id
if not cap_id:
cap_id = QUOTA_BYPASS_ALL if not (feature_id or "").strip() else quota_bypass_capability_id_for_feature(
feature_id.strip()
)
with get_db() as conn:
cur = get_cursor(conn)
cur.execute(
"""
DELETE FROM profile_capability_grants
WHERE profile_id = %s AND capability_id = %s
RETURNING profile_id, capability_id
""",
(profile_id, cap_id),
)
if not cur.fetchone():
raise HTTPException(status_code=404, detail="Grant nicht gefunden")
conn.commit()
return {"ok": True}

View File

@ -39,7 +39,7 @@ from ai_prompt_context import ExerciseFormAiFocusRow, ExerciseFormAiPromptContex
from ai_prompt_job import run_exercise_form_ai_suggestion
from account_lifecycle import assert_min_account_state
from capabilities import probe_capability
from club_features import probe_club_feature_access, resolve_club_id_for_probe
from club_features import consume_club_feature, probe_club_feature_access, resolve_club_id_for_probe
from exercise_rich_text import (
RICH_HTML_EXERCISE_FIELDS,
@ -2321,18 +2321,20 @@ def exercise_ai_suggest_endpoint(
OPENROUTER_API_KEY erforderlich.
"""
assert_min_account_state(tenant, "active_member", endpoint="POST /exercises/ai/suggest")
club_id = resolve_club_id_for_probe(tenant)
probe_capability(
tenant,
"exercises.ai.suggest",
action="suggest",
club_id=resolve_club_id_for_probe(tenant),
club_id=club_id,
endpoint="POST /exercises/ai/suggest",
)
probe_club_feature_access(
feature_id="ai_calls",
action="suggest",
club_id=resolve_club_id_for_probe(tenant),
club_id=club_id,
profile_id=tenant.profile_id,
portal_role=tenant.global_role,
endpoint="POST /exercises/ai/suggest",
)
with get_db() as conn:
@ -2344,6 +2346,14 @@ def exercise_ai_suggest_endpoint(
want_skills=body.include_skills,
want_instructions=body.include_instructions,
)
consume_club_feature(
feature_id="ai_calls",
club_id=club_id,
profile_id=tenant.profile_id,
portal_role=tenant.global_role,
action="suggest",
conn=conn,
)
return payload
@ -2355,18 +2365,20 @@ def exercise_ai_regenerate_endpoint(
):
"""Neu-Anfrage KI fuer eine gespeicherte Uebung; schreibendes Ergebnis nur im Frontend (PUT)."""
assert_min_account_state(tenant, "active_member", endpoint="POST /exercises/{id}/ai/regenerate")
club_id = resolve_club_id_for_probe(tenant)
probe_capability(
tenant,
"exercises.ai.regenerate",
action="regenerate",
club_id=resolve_club_id_for_probe(tenant),
club_id=club_id,
endpoint="POST /exercises/{id}/ai/regenerate",
)
probe_club_feature_access(
feature_id="ai_calls",
action="regenerate",
club_id=resolve_club_id_for_probe(tenant),
club_id=club_id,
profile_id=tenant.profile_id,
portal_role=tenant.global_role,
endpoint="POST /exercises/{id}/ai/regenerate",
)
want_summary = "summary" in body.regenerate
@ -2400,6 +2412,14 @@ def exercise_ai_regenerate_endpoint(
want_skills=want_skills,
want_instructions=want_instructions,
)
consume_club_feature(
feature_id="ai_calls",
club_id=club_id,
profile_id=tenant.profile_id,
portal_role=tenant.global_role,
action="regenerate",
conn=conn,
)
return payload
@ -2467,6 +2487,7 @@ def create_exercise(
action="create",
club_id=int(club_id),
profile_id=profile_id,
portal_role=tenant.global_role,
endpoint="POST /exercises",
)
@ -3285,6 +3306,7 @@ async def upload_exercise_media(
action="upload",
club_id=int(media_club_id),
profile_id=profile_id,
portal_role=tenant.global_role,
endpoint="POST /exercises/{id}/media",
conn=conn,
)

View File

@ -9,7 +9,7 @@ from planning_exercise_suggest import PlanningExerciseSuggestRequest, suggest_pl
from planning_exercise_path_builder import ProgressionPathSuggestRequest, suggest_progression_path
from account_lifecycle import assert_min_account_state
from capabilities import probe_capability
from club_features import probe_club_feature_access, resolve_club_id_for_probe
from club_features import consume_club_feature, probe_club_feature_access, resolve_club_id_for_probe
router = APIRouter(prefix="/api/planning", tags=["planning_exercise_suggest"])
@ -19,25 +19,38 @@ def post_planning_exercise_suggest(
body: PlanningExerciseSuggestRequest,
tenant: TenantContext = Depends(get_tenant_context),
):
if body.include_llm_intent or body.include_llm_rank:
uses_ai = body.include_llm_intent or body.include_llm_rank
club_id = resolve_club_id_for_probe(tenant) if uses_ai else None
if uses_ai:
assert_min_account_state(tenant, "active_member", endpoint="POST /planning/exercise-suggest")
probe_capability(
tenant,
"planning.ai.suggest",
action="planning_suggest",
club_id=resolve_club_id_for_probe(tenant),
club_id=club_id,
endpoint="POST /planning/exercise-suggest",
)
probe_club_feature_access(
feature_id="ai_calls",
action="planning_suggest",
club_id=resolve_club_id_for_probe(tenant),
club_id=club_id,
profile_id=tenant.profile_id,
portal_role=tenant.global_role,
endpoint="POST /planning/exercise-suggest",
)
with get_db() as conn:
cur = get_cursor(conn)
return suggest_planning_exercises(cur, tenant=tenant, body=body)
result = suggest_planning_exercises(cur, tenant=tenant, body=body)
if uses_ai:
consume_club_feature(
feature_id="ai_calls",
club_id=club_id,
profile_id=tenant.profile_id,
portal_role=tenant.global_role,
action="planning_suggest",
conn=conn,
)
return result
@router.post("/progression-path-suggest")
@ -45,11 +58,13 @@ def post_progression_path_suggest(
body: ProgressionPathSuggestRequest,
tenant: TenantContext = Depends(get_tenant_context),
):
if (
uses_ai = (
body.include_llm_intent
or body.include_llm_path_qa
or body.include_ai_gap_fill
):
)
club_id = resolve_club_id_for_probe(tenant) if uses_ai else None
if uses_ai:
assert_min_account_state(
tenant, "active_member", endpoint="POST /planning/progression-path-suggest"
)
@ -57,16 +72,27 @@ def post_progression_path_suggest(
tenant,
"planning.ai.progression_path",
action="progression_path_suggest",
club_id=resolve_club_id_for_probe(tenant),
club_id=club_id,
endpoint="POST /planning/progression-path-suggest",
)
probe_club_feature_access(
feature_id="ai_calls",
action="progression_path_suggest",
club_id=resolve_club_id_for_probe(tenant),
club_id=club_id,
profile_id=tenant.profile_id,
portal_role=tenant.global_role,
endpoint="POST /planning/progression-path-suggest",
)
with get_db() as conn:
cur = get_cursor(conn)
return suggest_progression_path(cur, tenant=tenant, body=body)
result = suggest_progression_path(cur, tenant=tenant, body=body)
if uses_ai:
consume_club_feature(
feature_id="ai_calls",
club_id=club_id,
profile_id=tenant.profile_id,
portal_role=tenant.global_role,
action="progression_path_suggest",
conn=conn,
)
return result

View File

@ -0,0 +1,138 @@
"""Kontingent-Bypass über Capability-Grants (kein Env-Hardcoding)."""
import pytest
from club_quota_bypass import (
QUOTA_BYPASS_ALL,
is_club_feature_quota_bypassed,
quota_bypass_access,
)
from club_features import consume_club_feature, probe_club_feature_access
class _FakeCur:
def __init__(self, *, portal_grants=None, profile_grants=None):
self._portal_grants = set(portal_grants or ())
self._profile_grants = set(profile_grants or ())
self._last_sql = ""
self._last_params = ()
def execute(self, sql, params=None):
self._last_sql = (sql or "").lower()
self._last_params = params or ()
def fetchone(self):
if "portal_role_capability_grants" in self._last_sql:
role, cap = self._last_params[:2]
if (role, cap) in self._portal_grants:
return {"1": 1}
if "profile_capability_grants" in self._last_sql:
pid, cap = self._last_params[:2]
if (int(pid), cap) in self._profile_grants:
return {"1": 1}
return None
def fetchall(self):
return []
def test_portal_role_grant_bypasses():
cur = _FakeCur(portal_grants={("superadmin", QUOTA_BYPASS_ALL)})
assert is_club_feature_quota_bypassed(
cur, profile_id=1, portal_role="superadmin", feature_id="ai_calls"
)
assert not is_club_feature_quota_bypassed(
cur, profile_id=1, portal_role="trainer", feature_id="ai_calls"
)
def test_profile_grant_bypasses():
cap = "platform.club_quota.bypass.ai_calls"
cur = _FakeCur(profile_grants={(42, cap)})
assert is_club_feature_quota_bypassed(
cur, profile_id=42, portal_role="trainer", feature_id="ai_calls"
)
def test_probe_superadmin_bypasses_enforce(monkeypatch):
monkeypatch.setenv("CLUB_FEATURE_ENFORCE", "1")
monkeypatch.setattr("club_feature_logger.log_club_feature_usage", lambda **kwargs: None)
class Cur:
def execute(self, *a, **k):
pass
def fetchone(self):
return {"1": 1}
def fetchall(self):
return []
monkeypatch.setattr("club_features.get_cursor", lambda c: Cur())
monkeypatch.setattr(
"club_features.get_effective_club_plan",
lambda cur, club_id: "free",
)
monkeypatch.setattr(
"club_quota_bypass._portal_role_has_grant",
lambda cur, role, cap: role == "superadmin",
)
monkeypatch.setattr(
"club_quota_bypass._profile_has_grant",
lambda cur, pid, cap: False,
)
access = probe_club_feature_access(
feature_id="ai_calls",
action="suggest",
club_id=5,
profile_id=1,
portal_role="superadmin",
conn=object(),
)
assert access["allowed"] is True
assert access["reason"] == "capability_quota_bypass"
def test_consume_skips_for_bypass_grant(monkeypatch):
calls = []
monkeypatch.setattr(
"club_features.increment_club_feature_usage",
lambda *a, **k: calls.append(1),
)
class Cur:
def execute(self, *a, **k):
pass
def fetchone(self):
return {"1": 1}
def fetchall(self):
return []
monkeypatch.setattr("club_features.get_cursor", lambda c: Cur())
monkeypatch.setattr(
"club_quota_bypass._portal_role_has_grant",
lambda cur, role, cap: role == "superadmin",
)
monkeypatch.setattr(
"club_quota_bypass._profile_has_grant",
lambda cur, pid, cap: False,
)
consume_club_feature(
feature_id="ai_calls",
club_id=9,
profile_id=1,
portal_role="superadmin",
conn=object(),
)
assert calls == []
def test_quota_bypass_access_shape():
row = quota_bypass_access(feature_id="ai_calls", club_id=3, plan_id="free")
assert row["platform_exempt"] is True
assert row["limit"] is None
assert row["allowed"] is True
assert row["reason"] == "capability_quota_bypass"

View File

@ -0,0 +1,122 @@
"""M5: ai_calls Verbrauch + Hard-Block (CLUB_FEATURE_ENFORCE)."""
import pytest
from fastapi import HTTPException
from club_features import (
club_feature_enforcement_enabled,
consume_club_feature,
probe_club_feature_access,
)
def _fake_cur():
class C:
def execute(self, *a, **k):
pass
def fetchone(self):
return None
return C()
def test_probe_blocks_when_enforce_and_limit_exceeded(monkeypatch):
monkeypatch.setenv("CLUB_FEATURE_ENFORCE", "1")
monkeypatch.setattr("club_features.get_cursor", lambda conn: _fake_cur())
monkeypatch.setattr(
"club_quota_bypass.is_club_feature_quota_bypassed",
lambda *a, **k: False,
)
monkeypatch.setattr(
"club_features.check_club_feature_access",
lambda club_id, feature_id, conn=None: {
"allowed": False,
"limit": 0,
"used": 0,
"remaining": 0,
"reason": "feature_disabled",
"plan_id": "free",
},
)
monkeypatch.setattr("club_feature_logger.log_club_feature_usage", lambda **kwargs: None)
with pytest.raises(HTTPException) as exc:
probe_club_feature_access(
feature_id="ai_calls",
action="suggest",
club_id=12,
profile_id=3,
endpoint="POST /exercises/ai/suggest",
conn=object(),
)
assert exc.value.status_code == 403
assert "ai_calls" in str(exc.value.detail)
def test_probe_allows_when_enforce_off(monkeypatch):
monkeypatch.setenv("CLUB_FEATURE_ENFORCE", "0")
monkeypatch.setattr("club_features.get_cursor", lambda conn: _fake_cur())
monkeypatch.setattr(
"club_quota_bypass.is_club_feature_quota_bypassed",
lambda *a, **k: False,
)
monkeypatch.setattr(
"club_features.check_club_feature_access",
lambda club_id, feature_id, conn=None: {
"allowed": False,
"limit": 0,
"used": 0,
"remaining": 0,
"reason": "feature_disabled",
"plan_id": "free",
},
)
monkeypatch.setattr("club_feature_logger.log_club_feature_usage", lambda **kwargs: None)
access = probe_club_feature_access(
feature_id="ai_calls",
action="suggest",
club_id=12,
profile_id=3,
conn=object(),
)
assert access["allowed"] is False
def test_consume_skips_without_club_id(monkeypatch):
calls = []
def _inc(*args, **kwargs):
calls.append(1)
monkeypatch.setattr("club_features.increment_club_feature_usage", _inc)
consume_club_feature(feature_id="ai_calls", club_id=None, profile_id=1)
assert calls == []
def test_consume_increments_once_per_call(monkeypatch):
calls = []
def _inc(club_id, feature_id, **kwargs):
calls.append((club_id, feature_id, kwargs.get("action")))
monkeypatch.setattr("club_features.get_cursor", lambda conn: _fake_cur())
monkeypatch.setattr(
"club_quota_bypass.is_club_feature_quota_bypassed",
lambda *a, **k: False,
)
monkeypatch.setattr("club_features.increment_club_feature_usage", _inc)
consume_club_feature(
feature_id="ai_calls",
club_id=5,
profile_id=9,
portal_role="trainer",
action="suggest",
conn=object(),
)
assert calls == [(5, "ai_calls", "suggest")]
def test_club_feature_enforcement_env_default_off(monkeypatch):
monkeypatch.delenv("CLUB_FEATURE_ENFORCE", raising=False)
assert club_feature_enforcement_enabled() is False

View File

@ -58,6 +58,10 @@ def test_build_me_entitlements_with_club(monkeypatch):
},
)
monkeypatch.setattr("entitlements._club_exists", lambda cur, cid: True)
monkeypatch.setattr(
"entitlements.is_club_feature_quota_bypassed",
lambda *a, **k: False,
)
tenant = TenantContext(
profile_id=3,

View File

@ -1,23 +1,24 @@
# Shinkan Jinkendo Version Information
APP_VERSION = "0.8.192"
BUILD_DATE = "2026-06-06"
DB_SCHEMA_VERSION = "20260606081"
APP_VERSION = "0.8.195"
BUILD_DATE = "2026-06-07"
DB_SCHEMA_VERSION = "20260606083"
MODULE_VERSIONS = {
"legal_documents": "1.4.0", # Admin: Live-Vorschau pro Abschnitt + modale Vollvorschau (Editor + Dokumentenliste)
"auth": "1.2.3", # P-05b: reset-password min_length=8 via Pydantic PasswordResetConfirm
"profiles": "1.8.1", # GET /profiles/me: account_state + club_roles
"tenant_context": "1.1.0", # M3: account_state + email_verified im TenantContext
"capabilities": "1.0.1", # resolve_capabilities_map für /me/entitlements
"capabilities": "1.1.0", # quota_bypass-Domain + profile_capability_grants in check_capability
"account_lifecycle": "1.1.0", # Phase A: account_onboarding_gate API-Middleware
"clubs": "0.4.2", # delete_club: Gründungsanträge → superseded
"club_memberships": "1.0.1", # Depends(get_tenant_context)
"club_join_requests": "1.0.1", # Depends(get_tenant_context)
"club_creation_requests": "1.0.1", # superseded wenn freigegebener Verein gelöscht
"admin_users": "1.0.0", # GET /api/admin/users
"club_features": "1.2.0", # M4: club_features_map für /me/entitlements
"entitlements": "1.0.0", # GET /api/me/entitlements — capabilities + features
"club_features": "1.5.0", # Kontingent-Bypass via Capability-Grants (probe/consume)
"club_quota_bypass": "1.0.0", # platform.club_quota.bypass* + Admin-Grants-API
"entitlements": "1.2.0", # capability_quota_bypass in Feature-Map für /me/entitlements
"platform_media_storage": "1.0.0", # GET/PUT /api/admin/platform-media-storage (Superadmin-Pfad unter MEDIA_ROOT)
"media_rights": "1.3.1", # acting_profile_id in write_audit_log_entry auf Optional[int] (P-13 anonyme Meldungen)
"media_assets": "1.18.1", # P-13: open_report_count in Listendaten (fuer Admins)
@ -34,7 +35,7 @@ MODULE_VERSIONS = {
"skill_profiles": "1.0.0", # Phase 3: gewichtetes Fähigkeiten-Profil + skill-discovery/suggestions
"methods": "0.1.0",
"exercises": "2.37.0", # Planungs-KI P1: Szenario-Pipeline + Query-Intent-Overlay
"planning_exercise_suggest": "0.16.0", # E3: gap_fill_offers, Off-Topic, QA→KI-Pipeline
"planning_exercise_suggest": "0.16.1", # M5: consume_club_feature ai_calls nach KI-Erfolg
"training_units": "0.4.0", # POST .../publish-to-framework: Ablauf aus geplanter Einheit → Rahmen-Slot-Blueprint
"training_programs": "0.1.0",
"planning": "0.15.0", # Vorlagen: Strukturvorschau, Bearbeiten inkl. Split-Sessions + Beschreibung

View File

@ -38,6 +38,8 @@ services:
APP_URL: "${APP_URL:-https://dev.shinkan.jinkendo.de}"
ALLOWED_ORIGINS: "${ALLOWED_ORIGINS:-https://dev.shinkan.jinkendo.de,http://192.168.2.49:3098}"
ENVIRONMENT: "${ENVIRONMENT:-development}"
# M5: Hard-Block Vereins-Kontingente (Default aus — in .env auf 1 setzen zum Testen)
CLUB_FEATURE_ENFORCE: "${CLUB_FEATURE_ENFORCE:-0}"
MEDIAWIKI_API_URL: "${MEDIAWIKI_API_URL:-https://karatetrainer.net/api.php}"
MEDIAWIKI_USER: "${MEDIAWIKI_USER:-Jinkendo}"
MEDIAWIKI_PASSWORD: "${MEDIAWIKI_PASSWORD:-CHANGE_ME}"

View File

@ -31,7 +31,20 @@ export default function FeatureUsageBadge({ featureId = 'ai_calls', label = 'KI-
return null
}
const { used = 0, limit, remaining, allowed } = feat
const { used = 0, limit, remaining, allowed, platform_exempt: platformExempt, reason } = feat
if (platformExempt || reason === 'platform_exempt' || reason === 'capability_quota_bypass') {
return (
<span
className="feature-usage-badge"
style={{ fontSize: '0.8rem', color: 'var(--accent-dark)' }}
title="Plattform-Ausnahme: zählt nicht gegen das Vereins-Kontingent"
>
{label}: Plattform (unbegrenzt)
</span>
)
}
// limit === 0 (z. B. Free-Plan ai_calls) anzeigen; nur echtes Unbegrenzt (null) ausblenden
if (limit == null) return null