b8fd99a5f4
- Service `app/services/test_templates.py`: CRUD with MITRE tag resolution (kind, external_id) → polymorphic join, filters by tactic/technique/ subtechnique/opsec/tag, `_UNSET` sentinel for partial-update semantics. - Service `app/services/scenario_templates.py`: ordered test list, reorder via full-replace (atomic w.r.t. UNIQUE(position) constraint), soft-delete. - REST endpoints on /api/v1/test-templates and /scenario-templates with pydantic schemas + perm gating (test_template.* and scenario_template.*). - /diag/reset truncates the 4 new tables before MITRE (FK ordering). - 19 pytest covering CRUD, MITRE tag merge, soft-delete chaining, perm enforcement, and reorder atomicity. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
116 lines
4.5 KiB
Python
116 lines
4.5 KiB
Python
"""Operational diagnostics. No auth in v1 (M0/M1 only expose non-sensitive
|
|
counts and the current Alembic revision).
|
|
|
|
The `/diag/reset` endpoint is **test-only** — it requires `APP_ENV=test` and
|
|
is the bedrock of the e2e suite (clean DB + freshly minted install token).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from flask import Blueprint, abort, jsonify
|
|
from sqlalchemy import text
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
|
|
from app.core.config import settings
|
|
from app.core.install_token import regenerate_install_token
|
|
from app.db.session import get_engine
|
|
|
|
bp = Blueprint("diag", __name__, url_prefix="/diag")
|
|
log = logging.getLogger("metamorph.diag")
|
|
|
|
|
|
@bp.get("/db")
|
|
def db_diag():
|
|
"""Return the Alembic revision and the count of public-schema tables."""
|
|
try:
|
|
with get_engine().connect() as conn:
|
|
revision = conn.execute(
|
|
text("SELECT version_num FROM alembic_version")
|
|
).scalar()
|
|
table_count = conn.execute(
|
|
text(
|
|
"SELECT count(*) FROM information_schema.tables "
|
|
"WHERE table_schema='public' AND table_type='BASE TABLE'"
|
|
)
|
|
).scalar_one()
|
|
except SQLAlchemyError as e:
|
|
log.warning("metamorph.diag.db_unreachable", extra={"error": str(e)})
|
|
return jsonify({"reachable": False, "error": "database_unreachable"}), 503
|
|
|
|
return jsonify(
|
|
{
|
|
"reachable": True,
|
|
"alembic_revision": revision,
|
|
"table_count": int(table_count),
|
|
}
|
|
)
|
|
|
|
|
|
@bp.post("/reset")
|
|
def reset_test_state():
|
|
"""TEST-ONLY: wipe users/auth tables and mint a fresh install token.
|
|
|
|
Refuses unless `APP_ENV=test`. Used by the Playwright suite to start each
|
|
auth scenario from a deterministic state.
|
|
"""
|
|
# NOTE: this endpoint is the test-suite reset hook. Allowed in `dev` too so
|
|
# the e2e suite can run against a normal `make up` stack, but in dev it is
|
|
# destructive — equivalent to `make clean` for the auth tables. Production
|
|
# (APP_ENV=prod/staging) is locked out.
|
|
if settings.APP_ENV not in ("dev", "test"):
|
|
abort(403, description="diag/reset is only available in dev/test")
|
|
if settings.APP_ENV == "dev":
|
|
log.warning("metamorph.diag.reset_in_dev_environment")
|
|
|
|
try:
|
|
with get_engine().begin() as conn:
|
|
# Auth + RBAC + settings reset.
|
|
conn.execute(
|
|
text(
|
|
"TRUNCATE users, refresh_tokens, invitations, invitation_groups, "
|
|
"user_groups, settings, groups RESTART IDENTITY CASCADE"
|
|
)
|
|
)
|
|
# Template catalogue reset (M5). The MITRE truncate below cascades to
|
|
# the polymorphic tag join, but the template rows themselves must be
|
|
# wiped first because `scenario_template_tests.test_template_id` is
|
|
# ON DELETE RESTRICT.
|
|
conn.execute(
|
|
text(
|
|
"TRUNCATE scenario_template_tests, scenario_templates, "
|
|
"test_template_mitre_tags, test_templates "
|
|
"RESTART IDENTITY CASCADE"
|
|
)
|
|
)
|
|
# MITRE reference reset — kept in sync with `settings` so a freshly
|
|
# reset stack has `GET /mitre/status` and `GET /mitre/tactics` agree
|
|
# ("no data, no last_sync"). The e2e suite re-syncs via /mitre/sync
|
|
# when it needs catalogue data.
|
|
conn.execute(
|
|
text(
|
|
"TRUNCATE mitre_technique_tactics, mitre_subtechniques, "
|
|
"mitre_techniques, mitre_tactics RESTART IDENTITY CASCADE"
|
|
)
|
|
)
|
|
except SQLAlchemyError as e:
|
|
log.error("metamorph.diag.reset_failed", extra={"error": str(e)})
|
|
return jsonify({"reset": False, "error": "database_error"}), 500
|
|
|
|
token = regenerate_install_token()
|
|
|
|
# Clear the in-memory rate-limit counters so the e2e suite that follows can
|
|
# log in repeatedly without hitting `/auth/login`/`/auth/refresh` limits.
|
|
# The limiter uses `memory://` in dev (cf. `app/core/rate_limit.py`).
|
|
try:
|
|
from app.core.rate_limit import limiter # noqa: PLC0415 — avoid import cycle
|
|
|
|
if limiter.enabled:
|
|
limiter.reset()
|
|
except Exception as e: # noqa: BLE001
|
|
log.warning("metamorph.diag.rate_limit_reset_failed", extra={"error": str(e)})
|
|
|
|
log.warning("metamorph.diag.reset_completed")
|
|
return jsonify({"reset": True, "install_token": token})
|