feat(data-safety): scheduled DB backups + audit trail + verified actor attribution

- db-backup sidecar: pg_dump -Fc of procurement+casdoor every
  BACKUP_INTERVAL_SECONDS into BACKUP_DIR, retention via BACKUP_KEEP_DAYS
- audit_log table (migration.sql + runtime ensure_audit_table) with
  model + transactional helper (utils/audit.record) — wired into user
  create/role-change and inventory transaction creation
- get_actor: quote_events.actor now stores the VERIFIED user's display
  name instead of the raw bearer token
This commit is contained in:
MMD
2026-09-18 22:23:34 +03:30
parent e17d1f0883
commit e5d6c77c3c
12 changed files with 187 additions and 14 deletions
+5
View File
@@ -40,6 +40,11 @@ DEV_LOGIN_ENABLED=false
# --- Deploy webhook (Gitea push → rebuild; see infra/deploy-webhook/) --- # --- Deploy webhook (Gitea push → rebuild; see infra/deploy-webhook/) ---
GITEA_WEBHOOK_SECRET=CHANGE_ME_long_random_webhook_secret GITEA_WEBHOOK_SECRET=CHANGE_ME_long_random_webhook_secret
# --- Backups (db-backup sidecar; see docker-compose.yml) ---
BACKUP_DIR=/srv/erp/backups
BACKUP_INTERVAL_SECONDS=86400
BACKUP_KEEP_DAYS=14
# --- Monitoring stack (see monitoring/README.md) --- # --- Monitoring stack (see monitoring/README.md) ---
# API log format for Loki ingestion ("text" locally, "json" in production) # API log format for Loki ingestion ("text" locally, "json" in production)
LOG_FORMAT=json LOG_FORMAT=json
+24
View File
@@ -392,3 +392,27 @@ async def ensure_v26(engine: AsyncEngine) -> None:
AFTER INSERT OR UPDATE OF status ON item_decisions AFTER INSERT OR UPDATE OF status ON item_decisions
FOR EACH ROW EXECUTE FUNCTION fn_sync_last_quoted_price() FOR EACH ROW EXECUTE FUNCTION fn_sync_last_quoted_price()
""") """)
async def ensure_audit_table(engine: AsyncEngine) -> None:
"""v2.7: general activity trail (audit_log). Idempotent; safe on any volume."""
ddl = """
CREATE TABLE IF NOT EXISTS audit_log (
id UUID NOT NULL DEFAULT gen_random_uuid() PRIMARY KEY,
occurred_at TIMESTAMPTZ NOT NULL DEFAULT now(),
actor TEXT,
action TEXT NOT NULL,
entity TEXT NOT NULL,
entity_id TEXT,
details JSONB NOT NULL DEFAULT '{}'
)
"""
indexes = (
"CREATE INDEX IF NOT EXISTS idx_audit_log_entity ON audit_log(entity, entity_id)",
"CREATE INDEX IF NOT EXISTS idx_audit_log_occurred ON audit_log(occurred_at DESC)",
)
async with engine.begin() as conn:
await conn.execute(text(ddl))
for stmt in indexes:
await conn.execute(text(stmt))
logger.info("audit_log table ensured")
+2
View File
@@ -24,6 +24,7 @@ from app.config import settings
from app.db.engine import AsyncSessionLocal, engine from app.db.engine import AsyncSessionLocal, engine
from app.db.init_db import ( from app.db.init_db import (
ensure_attachment_table, ensure_attachment_table,
ensure_audit_table,
ensure_dashboard_tables, ensure_dashboard_tables,
ensure_material_topic_column, ensure_material_topic_column,
ensure_quote_projects, ensure_quote_projects,
@@ -100,6 +101,7 @@ async def lifespan(_app: FastAPI):
await ensure_material_topic_column(engine) await ensure_material_topic_column(engine)
await ensure_quote_projects(engine) await ensure_quote_projects(engine)
await ensure_v26(engine) await ensure_v26(engine)
await ensure_audit_table(engine)
task = asyncio.create_task(_auto_expire_loop()) if settings.AUTO_EXPIRE_ENABLED else None task = asyncio.create_task(_auto_expire_loop()) if settings.AUTO_EXPIRE_ENABLED else None
try: try:
yield yield
+1
View File
@@ -1,5 +1,6 @@
"""ORM models — import order matters only for registration in Base.metadata.""" """ORM models — import order matters only for registration in Base.metadata."""
from app.db.base import Base from app.db.base import Base
from app.models.audit import AuditLog
from app.models.catalog import ( from app.models.catalog import (
AttributeDefinition, AttributeDefinition,
Brand, Brand,
+19
View File
@@ -0,0 +1,19 @@
"""Audit log model — general activity trail (v2.7)."""
import uuid
from sqlalchemy import Column, DateTime, String, Text, func
from sqlalchemy.dialects.postgresql import JSONB, UUID
from app.db.base import Base
class AuditLog(Base):
__tablename__ = "audit_log"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
occurred_at = Column(DateTime(timezone=True), nullable=False, server_default=func.now())
actor = Column(Text, nullable=True) # display name of the authenticated user
action = Column(String(64), nullable=False) # e.g. "user.role_changed"
entity = Column(String(64), nullable=False) # e.g. "user", "inventory_tx"
entity_id = Column(Text, nullable=True)
details = Column(JSONB, nullable=False, server_default="{}")
+11
View File
@@ -43,6 +43,7 @@ from app.utils.errors import (
UnauthorizedError, UnauthorizedError,
ValidationError, ValidationError,
) )
from app.utils.audit import record as audit_record
from app.utils.pagination import paginate, sort_query from app.utils.pagination import paginate, sort_query
from app.utils.serialization import to_iso from app.utils.serialization import to_iso
@@ -494,6 +495,16 @@ async def create_transaction(
if dest.project_id != scope.project_id: if dest.project_id != scope.project_id:
raise ForbiddenError("انبار مقصد متعلق به پروژه شما نیست") raise ForbiddenError("انبار مقصد متعلق به پروژه شما نیست")
txs = await apply_transaction(db, payload, performed_by=scope.actor) txs = await apply_transaction(db, payload, performed_by=scope.actor)
await audit_record(
db,
actor=scope.actor,
action="inventory.transaction_created",
entity="inventory",
entity_id=str(txs[0].inventory_id) if txs else None,
details={"type": payload.transaction_type, "quantity": str(payload.quantity),
"warehouse_id": str(payload.warehouse_id),
"tx_ids": [str(tx.id) for tx in txs]},
)
return {"data": [transaction_to_dict(tx) for tx in txs]} return {"data": [transaction_to_dict(tx) for tx in txs]}
+6 -2
View File
@@ -21,6 +21,7 @@ from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from app.auth import UserContext, get_current_user, require_role from app.auth import UserContext, get_current_user, require_role
from app.utils.audit import record as audit_record
from app.config import settings from app.config import settings
from app.db.engine import get_db from app.db.engine import get_db
from app.models.procurement import Project, User from app.models.procurement import Project, User
@@ -168,7 +169,7 @@ async def list_users(
@router.post("", status_code=201) @router.post("", status_code=201)
async def create_user( async def create_user(
payload: UserCreate, payload: UserCreate,
_: UserContext = Depends(_org_manager), ctx: UserContext = Depends(_org_manager),
db: AsyncSession = Depends(get_db), db: AsyncSession = Depends(get_db),
): ):
"""Define a new user (manual identity — casdoor_id optional).""" """Define a new user (manual identity — casdoor_id optional)."""
@@ -185,6 +186,7 @@ async def create_user(
) )
db.add(user) db.add(user)
await db.flush() await db.flush()
await audit_record(db, actor=ctx.display_name, action="user.created", entity="user", entity_id=str(user.id), details={"role": payload.role})
return {"data": user_to_dict(user)} return {"data": user_to_dict(user)}
@@ -192,7 +194,7 @@ async def create_user(
async def update_user_role( async def update_user_role(
user_id, user_id,
payload: UserRoleUpdate, payload: UserRoleUpdate,
_: UserContext = Depends(require_role("admin")), ctx: UserContext = Depends(require_role("admin")),
db: AsyncSession = Depends(get_db), db: AsyncSession = Depends(get_db),
): ):
"""Set/clear a role — ADMIN ONLY (RBAC): procurement managers must not """Set/clear a role — ADMIN ONLY (RBAC): procurement managers must not
@@ -201,10 +203,12 @@ async def update_user_role(
user = await db.get(User, user_id) user = await db.get(User, user_id)
if user is None: if user is None:
raise NotFoundError("کاربر یافت نشد") raise NotFoundError("کاربر یافت نشد")
previous_role = user.role
if user.role == "pm" and payload.role != "pm": if user.role == "pm" and payload.role != "pm":
await _unassign_projects(db, user.id) await _unassign_projects(db, user.id)
user.role = payload.role user.role = payload.role
await db.flush() await db.flush()
await audit_record(db, actor=ctx.display_name, action="user.role_changed", entity="user", entity_id=str(user.id), details={"from": previous_role, "to": payload.role})
projects = await _user_projects(db, user) projects = await _user_projects(db, user)
return {"data": user_to_dict(user, projects)} return {"data": user_to_dict(user, projects)}
+41
View File
@@ -0,0 +1,41 @@
"""Audit helper — one row per sensitive mutation, written in the SAME
transaction as the change (rolled back together with it).
await record(db, actor=ctx.display_name, action="user.role_changed",
entity="user", entity_id=str(user.id),
details={"from": old_role, "to": user.role})
Never raises: an audit failure must not fail the business operation.
"""
import logging
from typing import Any
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.audit import AuditLog
logger = logging.getLogger("procurement.audit")
async def record(
db: AsyncSession,
*,
actor: str | None,
action: str,
entity: str,
entity_id: str | None = None,
details: dict[str, Any] | None = None,
) -> None:
try:
db.add(
AuditLog(
actor=actor,
action=action,
entity=entity,
entity_id=str(entity_id) if entity_id else None,
details=details or {},
)
)
await db.flush()
except Exception: # noqa: BLE001 — audit must never break the caller
logger.exception("failed to write audit_log row (%s/%s)", action, entity)
+29 -12
View File
@@ -1,14 +1,31 @@
"""Placeholder auth: no middleware logic, only a header read for `actor` attribution.""" """Attribution for quote_events.actor — the VERIFIED user, not the raw token.
from fastapi import Request
Historically this stored the bearer token string verbatim (unverifiable and
def get_actor(request: Request) -> str | None: a token leak into the database). All routes that use it now sit behind the
"""Extract the bearer token (if any) to attribute quote_events.actor. global JWT dependency, so the token is resolved to the authenticated user
and the display name is recorded instead.
This is the reserved placeholder from the contract — no authentication
or authorization is enforced yet.
""" """
auth = request.headers.get("Authorization") from fastapi import Depends, Request
if auth and auth.lower().startswith("bearer "): from sqlalchemy.ext.asyncio import AsyncSession
return auth[7:].strip() or None
return None from app.auth import get_optional_user
from app.db.engine import get_db
async def get_actor(request: Request, db: AsyncSession = Depends(get_db)) -> str | None:
"""Resolve the caller's display name from the Casdoor Bearer token."""
ctx = await get_optional_user(request, db)
if ctx is not None:
return ctx.display_name
# Dev-mode header fallback (only active when DEV_LOGIN_ENABLED) carries
# the name in X-User-Name / X-User-Name-B64 — keep attribution useful.
name = request.headers.get("x-user-name")
import base64
b64 = request.headers.get("x-user-name-b64")
if b64:
try:
name = base64.b64decode(b64.encode()).decode("utf-8")
except Exception:
pass
return name
+33
View File
@@ -131,6 +131,39 @@ services:
restart: "no" restart: "no"
# ==========================================================================
# Scheduled logical backups of BOTH databases (procurement + casdoor).
# Writes pg_dump -Fc archives to a bind mount (default ./backups, point
# BACKUP_DIR at /srv/erp/backups on the VPS) and prunes archives older
# than BACKUP_KEEP_DAYS. Runs on the same postgres:18 image already pulled.
# ==========================================================================
db-backup:
image: postgres:18
container_name: procurement-db-backup
depends_on:
postgres:
condition: service_healthy
environment:
PGPASSWORD: ${POSTGRES_PASSWORD:-postgres}
BACKUP_INTERVAL_SECONDS: ${BACKUP_INTERVAL_SECONDS:-86400}
BACKUP_KEEP_DAYS: ${BACKUP_KEEP_DAYS:-14}
volumes:
- ${BACKUP_DIR:-./backups}:/backups
entrypoint: >
/bin/sh -c "
echo 'db-backup: dumping every $${BACKUP_INTERVAL_SECONDS}s, keeping $${BACKUP_KEEP_DAYS} days';
while true; do
STAMP=$$(date +%Y%m%d-%H%M%S);
pg_dump -h postgres -U postgres -d procurement -Fc -f /backups/procurement-$${STAMP}.dump &&
echo \"db-backup: procurement -> /backups/procurement-$${STAMP}.dump\";
pg_dump -h postgres -U postgres -d casdoor -Fc -f /backups/casdoor-$${STAMP}.dump &&
echo \"db-backup: casdoor -> /backups/casdoor-$${STAMP}.dump\";
find /backups -name '*.dump' -mtime +$${BACKUP_KEEP_DAYS} -delete;
sleep $${BACKUP_INTERVAL_SECONDS};
done
"
restart: unless-stopped
# ========================================================================== # ==========================================================================
# v2.6 — Casdoor (OIDC identity provider) # v2.6 — Casdoor (OIDC identity provider)
# Console: http://localhost:8080 (admin / 123) # Console: http://localhost:8080 (admin / 123)
+6 -1
View File
@@ -37,7 +37,12 @@ LISTEN_PORT = int(os.environ.get("WEBHOOK_PORT", "9000"))
SECRET = os.environ.get("GITEA_WEBHOOK_SECRET", "") SECRET = os.environ.get("GITEA_WEBHOOK_SECRET", "")
DEPLOY_DIR = os.environ.get("DEPLOY_DIR", "/srv/erp") DEPLOY_DIR = os.environ.get("DEPLOY_DIR", "/srv/erp")
DEPLOY_BRANCH = os.environ.get("DEPLOY_BRANCH", "production") DEPLOY_BRANCH = os.environ.get("DEPLOY_BRANCH", "production")
HEALTH_URL = os.environ.get("DEPLOY_HEALTH_URL", "") # e.g. http://127.0.0.1:8000/health/ready # Post-deploy readiness gate: defaults to the public API host from .env
# (both DEPLOY_HEALTH_URL and API_HOST live in the same EnvironmentFile).
HEALTH_URL = (
os.environ.get("DEPLOY_HEALTH_URL")
or (f"https://{os.environ['API_HOST']}/health/ready" if os.environ.get("API_HOST") else "")
)
LOCK_FILE = os.environ.get("DEPLOY_LOCK", "/tmp/erp-deploy.lock") LOCK_FILE = os.environ.get("DEPLOY_LOCK", "/tmp/erp-deploy.lock")
logging.basicConfig( logging.basicConfig(
+11
View File
@@ -634,4 +634,15 @@ CREATE TRIGGER trg_item_decisions_sync_price
-- double-applied. Arrival auto-posting (POST /pm/receipts) goes through -- double-applied. Arrival auto-posting (POST /pm/receipts) goes through
-- the same service path and is idempotent via receipt_allocation_id. -- the same service path and is idempotent via receipt_allocation_id.
-- ---------- AUDIT LOG (general activity trail, v2.7) ----------
CREATE TABLE IF NOT EXISTS audit_log (
id UUID NOT NULL DEFAULT gen_random_uuid() PRIMARY KEY,
occurred_at TIMESTAMPTZ NOT NULL DEFAULT now(),
actor TEXT,
action TEXT NOT NULL,
entity TEXT NOT NULL,
entity_id TEXT,
details JSONB NOT NULL DEFAULT '{}'
);
COMMIT; COMMIT;