Add phase 5 secure credential vault
This commit is contained in:
A R R R Associates
@@ -0,0 +1,159 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import date, datetime, timezone
|
||||
|
||||
from fastapi import HTTPException, Request
|
||||
from sqlalchemy import or_, select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.core.security.passwords import verify_password
|
||||
from app.modules.clients.models import Client
|
||||
from app.modules.core.iam.models import User
|
||||
from app.modules.core.rbac.deps import get_user_roles
|
||||
from app.modules.credential_vault.crypto import decrypt_value, encrypt_value
|
||||
from app.modules.credential_vault.models import CredentialVaultAccessLog, CredentialVaultEntry, CredentialVaultVersion
|
||||
|
||||
MANAGER_ROLES = {"Firm Admin", "Partner", "Branch Manager"}
|
||||
DENIED_ROLES = {"Client", "Consultant"}
|
||||
|
||||
|
||||
def utcnow() -> datetime:
|
||||
return datetime.now(timezone.utc)
|
||||
|
||||
|
||||
def active_tenant_id(request: Request, user: User) -> int:
|
||||
return int(request.session.get("active_tenant_id") or user.tenant_id)
|
||||
|
||||
|
||||
def active_branch_id(request: Request, user: User) -> int | None:
|
||||
raw = request.session.get("active_branch_id")
|
||||
return int(raw) if raw not in (None, "", 0, "0") else None
|
||||
|
||||
|
||||
def role_names(db: Session, user: User) -> set[str]:
|
||||
return set(get_user_roles(db, user.id))
|
||||
|
||||
|
||||
def can_open_vault(db: Session, user: User) -> bool:
|
||||
roles = role_names(db, user)
|
||||
return bool(roles & MANAGER_ROLES) or "Staff" in roles or "Employee" in roles
|
||||
|
||||
|
||||
def can_manage_vault(db: Session, user: User) -> bool:
|
||||
return bool(role_names(db, user) & MANAGER_ROLES)
|
||||
|
||||
|
||||
def _allowed_ids(entry: CredentialVaultEntry) -> set[int]:
|
||||
result: set[int] = set()
|
||||
for raw in (entry.allowed_user_ids_csv or "").split(","):
|
||||
try:
|
||||
result.add(int(raw.strip()))
|
||||
except (TypeError, ValueError):
|
||||
pass
|
||||
return result
|
||||
|
||||
|
||||
def can_view_entry(db: Session, user: User, entry: CredentialVaultEntry, branch_id: int | None = None) -> bool:
|
||||
roles = role_names(db, user)
|
||||
if roles & DENIED_ROLES:
|
||||
return False
|
||||
if entry.tenant_id != user.tenant_id and "System Admin" not in roles:
|
||||
return False
|
||||
if "Firm Admin" in roles or "Partner" in roles:
|
||||
return True
|
||||
if "Branch Manager" in roles:
|
||||
return not entry.branch_id or entry.branch_id == (branch_id or user.branch_id)
|
||||
return entry.owner_user_id == user.id or user.id in _allowed_ids(entry)
|
||||
|
||||
|
||||
def list_visible_entries(db: Session, user: User, tenant_id: int, branch_id: int | None, include_archived: bool = False):
|
||||
q = select(CredentialVaultEntry, Client).outerjoin(Client, Client.id == CredentialVaultEntry.client_id).where(CredentialVaultEntry.tenant_id == tenant_id)
|
||||
if not include_archived:
|
||||
q = q.where(CredentialVaultEntry.status != "archived")
|
||||
if branch_id:
|
||||
q = q.where(or_(CredentialVaultEntry.branch_id.is_(None), CredentialVaultEntry.branch_id == branch_id))
|
||||
rows = db.execute(q.order_by(CredentialVaultEntry.rotation_due_on.asc().nullslast(), CredentialVaultEntry.title)).all()
|
||||
return [(entry, client) for entry, client in rows if can_view_entry(db, user, entry, branch_id)]
|
||||
|
||||
|
||||
def log_access(db: Session, request: Request, user: User, entry: CredentialVaultEntry | None, action: str, *, reason: str | None = None, fields: str | None = None, success: bool = True) -> None:
|
||||
db.add(CredentialVaultAccessLog(
|
||||
tenant_id=entry.tenant_id if entry else user.tenant_id,
|
||||
branch_id=entry.branch_id if entry else user.branch_id,
|
||||
entry_id=entry.id if entry else None,
|
||||
actor_user_id=user.id,
|
||||
action=action,
|
||||
reason=reason,
|
||||
fields_accessed_csv=fields,
|
||||
success=success,
|
||||
ip_address=request.client.host if request.client else None,
|
||||
user_agent=request.headers.get("user-agent"),
|
||||
))
|
||||
|
||||
|
||||
def create_entry(db: Session, *, tenant_id: int, branch_id: int | None, client_id: int | None, registration_id: int | None, title: str, category: str, portal_url: str | None, reference_number: str | None, username: str | None, secret: str, additional_secret: str | None, notes: str | None, sensitivity: str, expires_on: date | None, rotation_due_on: date | None, owner_user_id: int | None, allowed_user_ids_csv: str | None, actor_user_id: int) -> CredentialVaultEntry:
|
||||
if not secret.strip():
|
||||
raise HTTPException(400, "Password, token or secret is required.")
|
||||
entry = CredentialVaultEntry(
|
||||
tenant_id=tenant_id, branch_id=branch_id, client_id=client_id, registration_id=registration_id,
|
||||
title=title.strip(), category=category, portal_url=portal_url or None, reference_number=reference_number or None,
|
||||
username_encrypted=encrypt_value(tenant_id, username), secret_encrypted=encrypt_value(tenant_id, secret) or "",
|
||||
additional_secret_encrypted=encrypt_value(tenant_id, additional_secret), notes_encrypted=encrypt_value(tenant_id, notes),
|
||||
sensitivity=sensitivity, expires_on=expires_on, rotation_due_on=rotation_due_on, owner_user_id=owner_user_id,
|
||||
allowed_user_ids_csv=allowed_user_ids_csv or None, created_by_user_id=actor_user_id, updated_by_user_id=actor_user_id,
|
||||
)
|
||||
db.add(entry)
|
||||
db.flush()
|
||||
return entry
|
||||
|
||||
|
||||
def reveal_entry(db: Session, request: Request, user: User, entry: CredentialVaultEntry, current_password: str, reason: str) -> dict[str, str]:
|
||||
if not can_view_entry(db, user, entry, active_branch_id(request, user)):
|
||||
log_access(db, request, user, entry, "reveal", reason=reason, success=False); db.commit()
|
||||
raise HTTPException(403, "You are not authorised to reveal this credential.")
|
||||
if not reason.strip():
|
||||
raise HTTPException(400, "A business reason is required.")
|
||||
if not verify_password(current_password, user.password_hash):
|
||||
log_access(db, request, user, entry, "reveal", reason=reason, success=False); db.commit()
|
||||
raise HTTPException(403, "Current ERP password is incorrect.")
|
||||
values = {
|
||||
"username": decrypt_value(entry.tenant_id, entry.username_encrypted),
|
||||
"secret": decrypt_value(entry.tenant_id, entry.secret_encrypted),
|
||||
"additional_secret": decrypt_value(entry.tenant_id, entry.additional_secret_encrypted),
|
||||
"notes": decrypt_value(entry.tenant_id, entry.notes_encrypted),
|
||||
}
|
||||
log_access(db, request, user, entry, "reveal", reason=reason.strip(), fields="username,secret,additional_secret,notes", success=True)
|
||||
db.commit()
|
||||
return values
|
||||
|
||||
|
||||
def rotate_entry(db: Session, entry: CredentialVaultEntry, *, username: str | None, secret: str, additional_secret: str | None, notes: str | None, rotation_due_on: date | None, reason: str, actor_user_id: int) -> None:
|
||||
if not secret.strip() or not reason.strip():
|
||||
raise HTTPException(400, "New secret and rotation reason are required.")
|
||||
version_count = db.execute(select(CredentialVaultVersion).where(CredentialVaultVersion.entry_id == entry.id)).scalars().all()
|
||||
db.add(CredentialVaultVersion(
|
||||
tenant_id=entry.tenant_id, entry_id=entry.id, version_number=len(version_count) + 1,
|
||||
username_encrypted=entry.username_encrypted, secret_encrypted=entry.secret_encrypted,
|
||||
additional_secret_encrypted=entry.additional_secret_encrypted, notes_encrypted=entry.notes_encrypted,
|
||||
change_reason=reason.strip(), changed_by_user_id=actor_user_id,
|
||||
))
|
||||
entry.username_encrypted = encrypt_value(entry.tenant_id, username)
|
||||
entry.secret_encrypted = encrypt_value(entry.tenant_id, secret) or ""
|
||||
entry.additional_secret_encrypted = encrypt_value(entry.tenant_id, additional_secret)
|
||||
entry.notes_encrypted = encrypt_value(entry.tenant_id, notes)
|
||||
entry.rotation_due_on = rotation_due_on
|
||||
entry.last_rotated_at_utc = utcnow()
|
||||
entry.updated_by_user_id = actor_user_id
|
||||
|
||||
|
||||
def due_state(entry: CredentialVaultEntry) -> str:
|
||||
today = date.today()
|
||||
due = entry.rotation_due_on or entry.expires_on
|
||||
if not due:
|
||||
return "not_scheduled"
|
||||
days = (due - today).days
|
||||
if days < 0:
|
||||
return "overdue"
|
||||
if days <= 30:
|
||||
return "due_soon"
|
||||
return "current"
|
||||
Reference in New Issue
Block a user