feat(backend): add Flask app factory, audit writer, flat CRUD + CLI (B0.7)

- Flask app factory wires SQLAlchemy / Migrate / Login / SocketIO and
  registers every blueprint. /healthz smoke endpoint included.
- Pydantic 2 DTOs (request/response) for engagement / host / TTP /
  scenario aggregates with from_attributes=True conversion.
- Flat CRUD blueprints under /api/v1/:
  * engagements (list / create / get / put / delete-as-archive)
  * hosts (engagement-scoped CRUD)
  * library/ttps (CRUD; promote requires the lead-only TTP_PROMOTE)
  * scenarios + steps (F3 invariant enforced: host.c2_type must match
    scenario.c2_type at compose time, 400 otherwise).
- @require_perm guards every endpoint per the F11 matrix.
- audit/ writer is hash-chained from v1 (SHA-256 of canonical record
  plus previous hash). The SQL-level write-only role enforcement ships
  in the deploy playbook (idempotent grants run at migration time).
- mimic-cli (click): user create (seeds RT operator/lead with group
  membership), db dump / db restore (manual pg_dump/pg_restore, R-O1).

No orchestrator, no WebSocket, no report generation — those land after
PR1/PR2/PR3.
This commit is contained in:
knacky
2026-05-21 20:33:45 +02:00
parent 7f4ad85a68
commit 9fa4d61304
17 changed files with 919 additions and 0 deletions

View File

@@ -0,0 +1,21 @@
"""`mimic-cli` command-line interface (click)."""
from __future__ import annotations
import click
from mimic.cli.db import db_group
from mimic.cli.user import user_group
@click.group()
def cli() -> None:
"""Mimic command-line interface."""
cli.add_command(user_group, name="user")
cli.add_command(db_group, name="db")
if __name__ == "__main__":
cli()

View File

@@ -0,0 +1,71 @@
"""Database CLI: dump / restore stubs (R-O1)."""
from __future__ import annotations
import shlex
import subprocess
from pathlib import Path
from urllib.parse import urlparse
import click
from mimic.config import get_settings
@click.group(help="Database operations (manual dump/restore per R-O1).")
def db_group() -> None: ...
def _parse_dsn(dsn: str) -> tuple[str, str, str, str, str]:
parsed = urlparse(dsn)
return (
parsed.hostname or "localhost",
str(parsed.port or 5432),
parsed.username or "",
parsed.password or "",
(parsed.path or "/").lstrip("/"),
)
@db_group.command("dump")
@click.option("--out", "out_path", type=click.Path(dir_okay=False, path_type=Path), required=True)
def dump(out_path: Path) -> None:
"""Manual `pg_dump` of the configured DATABASE_URL."""
settings = get_settings()
host, port, user, password, dbname = _parse_dsn(settings.database_url)
out_path.parent.mkdir(parents=True, exist_ok=True)
cmd = [
"pg_dump",
"--format=custom",
f"--host={host}",
f"--port={port}",
f"--username={user}",
f"--dbname={dbname}",
f"--file={out_path}",
]
env = {"PGPASSWORD": password} if password else None
click.echo(f"running: {shlex.join(cmd)}")
subprocess.run(cmd, check=True, env=env) # noqa: S603
click.echo(f"dump written to {out_path}")
@db_group.command("restore")
@click.option("--file", "in_path", type=click.Path(exists=True, path_type=Path), required=True)
def restore(in_path: Path) -> None:
"""Manual `pg_restore` from a dump file."""
settings = get_settings()
host, port, user, password, dbname = _parse_dsn(settings.database_url)
cmd = [
"pg_restore",
"--clean",
"--if-exists",
f"--host={host}",
f"--port={port}",
f"--username={user}",
f"--dbname={dbname}",
str(in_path),
]
env = {"PGPASSWORD": password} if password else None
click.echo(f"running: {shlex.join(cmd)}")
subprocess.run(cmd, check=True, env=env) # noqa: S603
click.echo("restore complete")

View File

@@ -0,0 +1,52 @@
"""User-related CLI commands."""
from __future__ import annotations
import click
from mimic.app import create_app
from mimic.auth.password import hash_password
from mimic.db.models import Group, User, UserGroup
from mimic.db.types import UserType
from mimic.extensions import db
from mimic.rbac.matrix import GroupName
@click.group(help="Manage Mimic user accounts.")
def user_group() -> None: ...
@user_group.command("create")
@click.option("--email", required=True)
@click.option(
"--type",
"user_type",
type=click.Choice([u.value for u in UserType]),
required=True,
)
@click.option("--password", prompt=True, hide_input=True, confirmation_prompt=True)
@click.option("--display-name", default=None)
def create_user(email: str, user_type: str, password: str, display_name: str | None) -> None:
"""Create a local user (sprint 0: rt_operator or rt_lead)."""
app = create_app()
with app.app_context():
user = User(
email=email,
display_name=display_name,
type=UserType(user_type),
local_password_hash=hash_password(password),
)
db.session.add(user)
db.session.flush()
group_name = (
GroupName.RT_LEAD if user.type is UserType.RT_LEAD else GroupName.RT_OPERATOR
)
group = db.session.query(Group).filter_by(name=group_name.value).first()
if group is None:
raise click.ClickException(
f"group {group_name.value} not seeded — run alembic upgrade head first"
)
db.session.add(UserGroup(user_id=user.id, group_id=group.id, engagement_id=None))
db.session.commit()
click.echo(f"created user {email} ({user.id}) in group {group_name.value}")