Restrict registration credentials by portal type and simplify vault form
This commit is contained in:
@@ -15,10 +15,45 @@ 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
|
||||
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
|
||||
@@ -61,18 +96,26 @@ def new_entry(request: Request, client_id: int | None = None, registration_id: i
|
||||
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)
|
||||
if client_id and not any(int(c.id) == int(client_id) for c in clients):
|
||||
raise HTTPException(404, "Client was not found in this audit firm.")
|
||||
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,
|
||||
),
|
||||
)
|
||||
|
||||
@@ -92,6 +135,21 @@ async def save_new(request: Request, title: str=Form(...), category: str=Form("g
|
||||
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)
|
||||
|
||||
Reference in New Issue
Block a user