101 lines
3.8 KiB
Python
101 lines
3.8 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.db.session import get_db
|
|
from app.models.api_client import ApiClient
|
|
from app.repositories.companies_repo import CompaniesRepository
|
|
from app.repositories.modules_repo import ModulesRepository
|
|
from app.repositories.permissions_repo import PermissionsRepository
|
|
from app.repositories.sites_repo import SitesRepository
|
|
from app.repositories.systems_repo import SystemsRepository
|
|
from app.repositories.users_repo import UsersRepository
|
|
from app.schemas.permissions import PermissionGrantRequest, PermissionRevokeRequest
|
|
from app.security.api_client_auth import require_api_client
|
|
|
|
router = APIRouter(prefix="/admin", tags=["admin"])
|
|
|
|
|
|
def _resolve_module_id(db: Session, system_key: str, module_key: str | None) -> str:
|
|
systems_repo = SystemsRepository(db)
|
|
modules_repo = ModulesRepository(db)
|
|
system = systems_repo.get_by_key(system_key)
|
|
if not system:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="system_not_found")
|
|
|
|
target_module_key = f"{system_key}.{module_key}" if module_key else f"{system_key}.__system__"
|
|
module = modules_repo.get_by_key(target_module_key)
|
|
if not module:
|
|
module = modules_repo.create(module_key=target_module_key, name=target_module_key, status="active")
|
|
return module.id
|
|
|
|
|
|
def _resolve_scope_ids(db: Session, scope_type: str, scope_id: str) -> tuple[str | None, str | None]:
|
|
companies_repo = CompaniesRepository(db)
|
|
sites_repo = SitesRepository(db)
|
|
if scope_type == "company":
|
|
company = companies_repo.get_by_key(scope_id)
|
|
if not company:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="company_not_found")
|
|
return company.id, None
|
|
if scope_type == "site":
|
|
site = sites_repo.get_by_key(scope_id)
|
|
if not site:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="site_not_found")
|
|
return None, site.id
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="invalid_scope_type")
|
|
|
|
|
|
@router.post("/permissions/grant")
|
|
def grant_permission(
|
|
payload: PermissionGrantRequest,
|
|
_: ApiClient = Depends(require_api_client),
|
|
db: Session = Depends(get_db),
|
|
) -> dict[str, str]:
|
|
users_repo = UsersRepository(db)
|
|
perms_repo = PermissionsRepository(db)
|
|
|
|
user = users_repo.upsert_by_sub(
|
|
authentik_sub=payload.authentik_sub,
|
|
email=payload.email,
|
|
display_name=payload.display_name,
|
|
is_active=True,
|
|
)
|
|
module_id = _resolve_module_id(db, payload.system, payload.module)
|
|
company_id, site_id = _resolve_scope_ids(db, payload.scope_type, payload.scope_id)
|
|
permission = perms_repo.create_if_not_exists(
|
|
user_id=user.id,
|
|
module_id=module_id,
|
|
action=payload.action,
|
|
scope_type=payload.scope_type,
|
|
company_id=company_id,
|
|
site_id=site_id,
|
|
)
|
|
|
|
return {"permission_id": permission.id, "result": "granted"}
|
|
|
|
|
|
@router.post("/permissions/revoke")
|
|
def revoke_permission(
|
|
payload: PermissionRevokeRequest,
|
|
_: ApiClient = Depends(require_api_client),
|
|
db: Session = Depends(get_db),
|
|
) -> dict[str, int | str]:
|
|
users_repo = UsersRepository(db)
|
|
perms_repo = PermissionsRepository(db)
|
|
|
|
user = users_repo.get_by_sub(payload.authentik_sub)
|
|
if user is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="user_not_found")
|
|
|
|
module_id = _resolve_module_id(db, payload.system, payload.module)
|
|
company_id, site_id = _resolve_scope_ids(db, payload.scope_type, payload.scope_id)
|
|
deleted = perms_repo.revoke(
|
|
user_id=user.id,
|
|
module_id=module_id,
|
|
action=payload.action,
|
|
scope_type=payload.scope_type,
|
|
company_id=company_id,
|
|
site_id=site_id,
|
|
)
|
|
return {"deleted": deleted, "result": "revoked"}
|