A R R R Associates
201 lines
14 KiB
Python
201 lines
14 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, 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)
|