A R R R Associates
120 lines
8.9 KiB
Python
120 lines
8.9 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import date
|
|
|
|
from fastapi import APIRouter, Form, HTTPException, Request
|
|
from fastapi.responses import HTMLResponse, RedirectResponse
|
|
from sqlalchemy import select
|
|
|
|
from app.core.db.common import CommonSessionLocal
|
|
from app.core.security.csrf import get_or_create_csrf_token, validate_csrf
|
|
from app.core.security.session_auth import get_current_user
|
|
from app.core.templating import templates
|
|
from app.modules.clients.models import Client
|
|
from app.modules.core.iam.models import User
|
|
from app.modules.core.rbac.deps import get_user_permissions, get_user_roles
|
|
from app.modules.credential_vault.models import CredentialVaultAccessLog, CredentialVaultEntry, CredentialVaultVersion
|
|
from app.modules.credential_vault.service import active_branch_id, active_tenant_id, can_manage_vault, can_open_vault, can_view_entry, create_entry, due_state, list_visible_entries, log_access, reveal_entry, rotate_entry
|
|
from app.modules.registrations.models import ClientRegistration
|
|
|
|
router = APIRouter(prefix="/credential-vault", tags=["credential-vault-ui"])
|
|
|
|
|
|
def _date(value: str) -> date | None:
|
|
return date.fromisoformat(value) if value else None
|
|
|
|
|
|
def _ctx(request: Request, user: User, db, **kwargs):
|
|
data = {"request": request, "current_user": user, "current_user_roles": get_user_roles(db, user.id), "current_user_permissions": get_user_permissions(db, user.id), "csrf_token": get_or_create_csrf_token(request)}
|
|
data.update(kwargs)
|
|
return data
|
|
|
|
|
|
def _user(request: Request, db) -> User:
|
|
user = get_current_user(request, db)
|
|
if not user:
|
|
raise HTTPException(401, "Login required")
|
|
if not can_open_vault(db, user):
|
|
raise HTTPException(403, "Credential vault access is not enabled for this role.")
|
|
return user
|
|
|
|
|
|
@router.get("", response_class=HTMLResponse)
|
|
def dashboard(request: Request, include_archived: bool = False):
|
|
with CommonSessionLocal() as db:
|
|
user = _user(request, db); tenant_id = active_tenant_id(request, user); branch_id = active_branch_id(request, user)
|
|
rows = list_visible_entries(db, user, tenant_id, branch_id, include_archived)
|
|
counts = {"total": len(rows), "due_soon": 0, "overdue": 0}
|
|
decorated = []
|
|
for entry, client in rows:
|
|
state = due_state(entry); counts[state] = counts.get(state, 0) + 1; decorated.append((entry, client, state))
|
|
return templates.TemplateResponse("modules/credential_vault/templates/credential_vault/dashboard.html", _ctx(request, user, db, rows=decorated, counts=counts, can_manage=can_manage_vault(db, user), include_archived=include_archived))
|
|
|
|
|
|
@router.get("/new", response_class=HTMLResponse)
|
|
def new_entry(request: Request):
|
|
with CommonSessionLocal() as db:
|
|
user = _user(request, db)
|
|
if not can_manage_vault(db, user): raise HTTPException(403, "Only firm managers may create credentials.")
|
|
tenant_id = active_tenant_id(request, user); branch_id = active_branch_id(request, user)
|
|
clients = db.execute(select(Client).where(Client.tenant_id == tenant_id, Client.is_active.is_(True)).order_by(Client.client_name)).scalars().all()
|
|
users = db.execute(select(User).where(User.tenant_id == tenant_id, User.is_active.is_(True)).order_by(User.full_name)).scalars().all()
|
|
registrations = db.execute(select(ClientRegistration).where(ClientRegistration.tenant_id == tenant_id).order_by(ClientRegistration.registration_number)).scalars().all()
|
|
return templates.TemplateResponse("modules/credential_vault/templates/credential_vault/form.html", _ctx(request, user, db, entry=None, clients=clients, users=users, registrations=registrations, branch_id=branch_id))
|
|
|
|
|
|
@router.post("/new")
|
|
async def save_new(request: Request, title: str=Form(...), category: str=Form("government_portal"), client_id: str=Form(""), registration_id: str=Form(""), portal_url: str=Form(""), reference_number: str=Form(""), username: str=Form(""), secret: str=Form(...), additional_secret: str=Form(""), notes: str=Form(""), sensitivity: str=Form("high"), expires_on: str=Form(""), rotation_due_on: str=Form(""), owner_user_id: str=Form(""), allowed_user_ids: list[str]=Form(default=[]), csrf_token: str=Form(...)):
|
|
with CommonSessionLocal() as db:
|
|
user = _user(request, db); validate_csrf(request, csrf_token)
|
|
if not can_manage_vault(db, user): raise HTTPException(403)
|
|
tenant_id = active_tenant_id(request, user)
|
|
entry = create_entry(db, tenant_id=tenant_id, branch_id=active_branch_id(request, user), client_id=int(client_id) if client_id else None, registration_id=int(registration_id) if registration_id else None, title=title, category=category, portal_url=portal_url, reference_number=reference_number, username=username, secret=secret, additional_secret=additional_secret, notes=notes, sensitivity=sensitivity, expires_on=_date(expires_on), rotation_due_on=_date(rotation_due_on), owner_user_id=int(owner_user_id) if owner_user_id else user.id, allowed_user_ids_csv=",".join(allowed_user_ids), actor_user_id=user.id)
|
|
log_access(db, request, user, entry, "create", reason="Credential created"); db.commit()
|
|
return RedirectResponse(f"/credential-vault/{entry.id}", 303)
|
|
|
|
|
|
@router.get("/{entry_id}", response_class=HTMLResponse)
|
|
def detail(request: Request, entry_id: int):
|
|
with CommonSessionLocal() as db:
|
|
user = _user(request, db); entry = db.get(CredentialVaultEntry, entry_id)
|
|
if not entry or not can_view_entry(db, user, entry, active_branch_id(request, user)): raise HTTPException(404)
|
|
client = db.get(Client, entry.client_id) if entry.client_id else None
|
|
versions = db.execute(select(CredentialVaultVersion).where(CredentialVaultVersion.entry_id == entry.id).order_by(CredentialVaultVersion.version_number.desc())).scalars().all()
|
|
logs = db.execute(select(CredentialVaultAccessLog, User).outerjoin(User, User.id == CredentialVaultAccessLog.actor_user_id).where(CredentialVaultAccessLog.entry_id == entry.id).order_by(CredentialVaultAccessLog.created_at_utc.desc()).limit(100)).all()
|
|
return templates.TemplateResponse("modules/credential_vault/templates/credential_vault/detail.html", _ctx(request, user, db, entry=entry, client=client, versions=versions, logs=logs, due_state=due_state(entry), can_manage=can_manage_vault(db, user)))
|
|
|
|
|
|
@router.post("/{entry_id}/reveal", response_class=HTMLResponse)
|
|
async def reveal(request: Request, entry_id: int, current_password: str=Form(...), reason: str=Form(...), csrf_token: str=Form(...)):
|
|
with CommonSessionLocal() as db:
|
|
user = _user(request, db); validate_csrf(request, csrf_token); entry = db.get(CredentialVaultEntry, entry_id)
|
|
if not entry: raise HTTPException(404)
|
|
values = reveal_entry(db, request, user, entry, current_password, reason)
|
|
response = templates.TemplateResponse("modules/credential_vault/templates/credential_vault/reveal.html", _ctx(request, user, db, entry=entry, values=values, reason=reason))
|
|
response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate, private"; response.headers["Pragma"] = "no-cache"; response.headers["X-Content-Type-Options"] = "nosniff"
|
|
return response
|
|
|
|
|
|
@router.post("/{entry_id}/rotate")
|
|
async def rotate(request: Request, entry_id: int, username: str=Form(""), secret: str=Form(...), additional_secret: str=Form(""), notes: str=Form(""), rotation_due_on: str=Form(""), reason: str=Form(...), csrf_token: str=Form(...)):
|
|
with CommonSessionLocal() as db:
|
|
user = _user(request, db); validate_csrf(request, csrf_token)
|
|
if not can_manage_vault(db, user): raise HTTPException(403)
|
|
entry = db.get(CredentialVaultEntry, entry_id)
|
|
if not entry or not can_view_entry(db, user, entry, active_branch_id(request, user)): raise HTTPException(404)
|
|
rotate_entry(db, entry, username=username, secret=secret, additional_secret=additional_secret, notes=notes, rotation_due_on=_date(rotation_due_on), reason=reason, actor_user_id=user.id)
|
|
log_access(db, request, user, entry, "rotate", reason=reason); db.commit(); return RedirectResponse(f"/credential-vault/{entry.id}", 303)
|
|
|
|
|
|
@router.post("/{entry_id}/archive")
|
|
async def archive(request: Request, entry_id: int, reason: str=Form(...), csrf_token: str=Form(...)):
|
|
with CommonSessionLocal() as db:
|
|
user = _user(request, db); validate_csrf(request, csrf_token)
|
|
if not can_manage_vault(db, user): raise HTTPException(403)
|
|
entry = db.get(CredentialVaultEntry, entry_id)
|
|
if not entry or not can_view_entry(db, user, entry, active_branch_id(request, user)): raise HTTPException(404)
|
|
if not reason.strip(): raise HTTPException(400, "Archive reason is required.")
|
|
entry.status="archived"; entry.archived_at_utc=__import__('datetime').datetime.now(__import__('datetime').timezone.utc); entry.updated_by_user_id=user.id
|
|
log_access(db, request, user, entry, "archive", reason=reason); db.commit(); return RedirectResponse("/credential-vault", 303)
|