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, RegistrationType router = APIRouter(prefix="/credential-vault", tags=["credential-vault-ui"]) REGISTRATION_CREDENTIAL_TYPES: dict[str, tuple[tuple[str, str], ...]] = { "PAN": (("income_tax_portal", "Income Tax Portal"), ("other", "Other")), "TAN": (("income_tax_portal", "Income Tax Portal / TDS"), ("traces_tds", "TRACES / TDS"), ("other", "Other")), "GSTIN": (("gst_portal", "GST Portal"), ("eway_bill", "E-Way Bill"), ("einvoice", "E-Invoice"), ("api_key", "GST API / Provider"), ("other", "Other")), "CIN": (("mca_portal", "MCA Portal"), ("other", "Other")), "LLPIN": (("mca_portal", "MCA Portal"), ("other", "Other")), "DSC": (("digital_signature", "Digital Signature / Token"), ("other", "Other")), "UDYAM": (("government_portal", "Udyam / MSME Portal"), ("other", "Other")), "FSSAI": (("government_portal", "FSSAI Portal"), ("other", "Other")), } GENERIC_CREDENTIAL_TYPES: tuple[tuple[str, str], ...] = ( ("gst_portal", "GST Portal"), ("income_tax_portal", "Income Tax Portal"), ("traces_tds", "TRACES / TDS"), ("mca_portal", "MCA Portal"), ("eway_bill", "E-Way Bill"), ("einvoice", "E-Invoice"), ("government_portal", "Other Government Portal"), ("banking", "Banking"), ("email", "Email"), ("software", "Software"), ("api_key", "API key"), ("digital_signature", "Digital signature"), ("other", "Other"), ) def _credential_types_for_registration_type(type_code: str | None) -> tuple[tuple[str, str], ...]: code = (type_code or "").strip().upper() return REGISTRATION_CREDENTIAL_TYPES.get(code, (("government_portal", "Government / Registration Portal"), ("other", "Other"))) def _category_label(category: str, choices: tuple[tuple[str, str], ...]) -> str: return dict(choices).get(category, category.replace("_", " ").title()) 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, client_id: int | None = None, registration_id: int | None = None): 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() selected_registration = None selected_registration_type = None selected_client = None credential_types = GENERIC_CREDENTIAL_TYPES if registration_id: selected_registration = db.get(ClientRegistration, int(registration_id)) if not selected_registration or selected_registration.tenant_id != tenant_id: raise HTTPException(404, "Registration record was not found in this audit firm.") client_id = int(selected_registration.client_id) selected_registration_type = db.get(RegistrationType, selected_registration.registration_type_id) credential_types = _credential_types_for_registration_type(getattr(selected_registration_type, "code", None)) if client_id: selected_client = next((c for c in clients if int(c.id) == int(client_id)), None) if not selected_client: raise HTTPException(404, "Client was not found in this audit firm.") 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, selected_client_id=client_id, selected_registration_id=registration_id, selected_registration=selected_registration, selected_registration_type=selected_registration_type, selected_client=selected_client, credential_types=credential_types, ), ) @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) selected_client_id = int(client_id) if client_id else None selected_registration_id = int(registration_id) if registration_id else None if selected_registration_id: registration = db.get(ClientRegistration, selected_registration_id) if not registration or registration.tenant_id != tenant_id: raise HTTPException(400, "Selected registration is not available in this audit firm.") if selected_client_id and int(registration.client_id) != selected_client_id: raise HTTPException(400, "Selected registration does not belong to the selected client.") selected_client_id = int(registration.client_id) registration_type = db.get(RegistrationType, registration.registration_type_id) allowed_types = _credential_types_for_registration_type(getattr(registration_type, "code", None)) allowed_codes = {code for code, _label in allowed_types} if category not in allowed_codes: type_name = getattr(registration_type, "name", None) or getattr(registration_type, "code", None) or "this registration type" raise HTTPException(400, f"{_category_label(category, GENERIC_CREDENTIAL_TYPES)} credentials cannot be linked to {type_name}.") # Registration-linked credentials intentionally use the simpler office workflow. # Generic metadata columns remain available for firm-level vault entries. portal_url = "" reference_number = registration.registration_number or "" sensitivity = "high" expires_on = "" rotation_due_on = "" owner_user_id = str(user.id) allowed_user_ids = [] entry = create_entry(db, tenant_id=tenant_id, branch_id=active_branch_id(request, user), client_id=selected_client_id, registration_id=selected_registration_id, 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)