a04d5308ab
Some checks failed
Aegis CI / lint-and-test (push) Has been cancelled
- jira-test now returns {status: "ok"|"error", message: ...} with
HTTP 200 so Cloudflare never intercepts the response
- jira_service strips trailing slash from URL before creating Jira
client (avoids double-slash in REST paths)
- Frontend reads data.status field instead of HTTP status code
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
417 lines
13 KiB
Python
417 lines
13 KiB
Python
"""System-level endpoints (admin only).
|
|
|
|
Provides manual triggers for background operations such as the MITRE
|
|
ATT&CK synchronisation, intel scanning, Atomic Red Team import, and
|
|
scheduler health introspection.
|
|
|
|
Also exposes email configuration CRUD (admin only) that writes to the
|
|
system_configs table so settings survive container restarts.
|
|
"""
|
|
|
|
import logging
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request, status
|
|
from pydantic import BaseModel
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.database import SessionLocal, get_db
|
|
from app.dependencies.auth import require_role
|
|
from app.models.user import User
|
|
from app.services.mitre_sync_service import sync_mitre
|
|
from app.services.intel_service import scan_intel
|
|
from app.services.atomic_import_service import import_atomic_red_team
|
|
from app.jobs.mitre_sync_job import scheduler
|
|
from app.limiter import limiter
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/system", tags=["system"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Pydantic schemas for email config
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class EmailConfigOut(BaseModel):
|
|
enabled: bool
|
|
host: str
|
|
port: int
|
|
username: str
|
|
from_email: str
|
|
use_tls: bool
|
|
# password is never returned
|
|
|
|
|
|
class EmailConfigUpdate(BaseModel):
|
|
enabled: Optional[bool] = None
|
|
host: Optional[str] = None
|
|
port: Optional[int] = None
|
|
username: Optional[str] = None
|
|
password: Optional[str] = None
|
|
from_email: Optional[str] = None
|
|
use_tls: Optional[bool] = None
|
|
|
|
|
|
class EmailTestRequest(BaseModel):
|
|
to: str
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers for system_configs CRUD
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_SMTP_KEYS = {
|
|
"enabled": "smtp.enabled",
|
|
"host": "smtp.host",
|
|
"port": "smtp.port",
|
|
"username": "smtp.username",
|
|
"password": "smtp.password",
|
|
"from_email": "smtp.from_email",
|
|
"use_tls": "smtp.use_tls",
|
|
}
|
|
|
|
|
|
def _upsert_config(db: Session, key: str, value: str) -> None:
|
|
from app.models.system_config import SystemConfig # lazy import avoids circular
|
|
|
|
row = db.query(SystemConfig).filter(SystemConfig.key == key).first()
|
|
if row:
|
|
row.value = value
|
|
else:
|
|
row = SystemConfig(key=key, value=value)
|
|
db.add(row)
|
|
|
|
|
|
def _read_email_config_from_db(db: Session) -> dict:
|
|
"""Return a dict with resolved email settings (DB overrides env)."""
|
|
from app.services.email_service import _get_smtp_config
|
|
|
|
return _get_smtp_config(db)
|
|
|
|
|
|
def _bg_mitre_sync() -> None:
|
|
"""Run MITRE sync in a background task with its own DB session."""
|
|
logger.info("Background MITRE sync task starting...")
|
|
db = SessionLocal()
|
|
try:
|
|
summary = sync_mitre(db)
|
|
logger.info("Background MITRE sync task finished — %s", summary)
|
|
except Exception:
|
|
logger.exception("Background MITRE sync task failed")
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/sync-mitre")
|
|
@limiter.limit("2/hour")
|
|
def trigger_mitre_sync(
|
|
request: Request,
|
|
background_tasks: BackgroundTasks,
|
|
current_user: User = Depends(require_role("admin")),
|
|
):
|
|
"""Manually trigger a MITRE ATT&CK synchronisation in the background.
|
|
|
|
**Requires** the ``admin`` role.
|
|
|
|
Returns immediately — the sync runs asynchronously. Poll
|
|
``/system/scheduler-status`` for progress, or check server logs.
|
|
"""
|
|
background_tasks.add_task(_bg_mitre_sync)
|
|
return {
|
|
"message": "MITRE sync started in background",
|
|
"status": "started",
|
|
"new": 0,
|
|
"updated": 0,
|
|
}
|
|
|
|
|
|
@router.post("/run-intel-scan")
|
|
def trigger_intel_scan(
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_role("admin")),
|
|
):
|
|
"""Manually trigger a threat-intelligence scan.
|
|
|
|
**Requires** the ``admin`` role.
|
|
|
|
Returns a JSON object with the scan summary including the count of
|
|
new intel items found.
|
|
"""
|
|
summary = scan_intel(db)
|
|
return {
|
|
"message": "Intel scan completed",
|
|
"new_items": summary["new_items"],
|
|
}
|
|
|
|
|
|
@router.post("/import-atomic-tests")
|
|
@limiter.limit("2/hour")
|
|
def trigger_atomic_import(
|
|
request: Request,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_role("admin")),
|
|
):
|
|
"""Trigger an import of Atomic Red Team tests as TestTemplates.
|
|
|
|
**Requires** the ``admin`` role.
|
|
|
|
Downloads the Atomic Red Team repository ZIP from GitHub, parses the
|
|
YAML files, and creates/updates TestTemplate records. Running this
|
|
endpoint multiple times is idempotent — duplicates are skipped.
|
|
|
|
Returns a JSON object with import statistics.
|
|
"""
|
|
try:
|
|
summary = import_atomic_red_team(db)
|
|
except Exception as exc:
|
|
logger.error("Atomic Red Team import failed: %s", exc, exc_info=True)
|
|
return {
|
|
"message": "Import failed. Check server logs for details.",
|
|
}
|
|
|
|
return {
|
|
"message": "Import completed",
|
|
"imported": summary["created"],
|
|
"skipped": summary["skipped_existing"],
|
|
"total_parsed": summary["total_tests_parsed"],
|
|
}
|
|
|
|
|
|
@router.get("/scheduler-status")
|
|
def scheduler_status(
|
|
current_user: User = Depends(require_role("admin")),
|
|
):
|
|
"""Return the current state of the background scheduler.
|
|
|
|
**Requires** the ``admin`` role.
|
|
"""
|
|
jobs = scheduler.get_jobs()
|
|
return {
|
|
"running": scheduler.running,
|
|
"jobs": [
|
|
{
|
|
"id": job.id,
|
|
"name": job.name,
|
|
"next_run_time": str(job.next_run_time) if job.next_run_time else None,
|
|
}
|
|
for job in jobs
|
|
],
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Jira config endpoints (admin only)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class JiraConfigOut(BaseModel):
|
|
enabled: bool
|
|
url: str
|
|
project_key: str
|
|
parent_ticket: str
|
|
# Credentials are never returned
|
|
|
|
|
|
class JiraConfigUpdate(BaseModel):
|
|
enabled: Optional[bool] = None
|
|
url: Optional[str] = None
|
|
project_key: Optional[str] = None
|
|
parent_ticket: Optional[str] = None
|
|
|
|
|
|
_JIRA_KEYS = {
|
|
"enabled": "jira.enabled",
|
|
"url": "jira.url",
|
|
"project_key": "jira.project_key",
|
|
"parent_ticket": "jira.parent_ticket",
|
|
}
|
|
|
|
|
|
@router.get("/jira-config", response_model=JiraConfigOut)
|
|
def get_jira_config(
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_role("admin")),
|
|
):
|
|
"""Return current Jira configuration (merged DB + env).
|
|
|
|
**Requires** the ``admin`` role. Credentials are never returned.
|
|
"""
|
|
from app.services.jira_service import get_jira_url, get_jira_project_key, is_jira_enabled, get_jira_parent_ticket
|
|
|
|
return JiraConfigOut(
|
|
enabled=is_jira_enabled(db),
|
|
url=get_jira_url(db) or "",
|
|
project_key=get_jira_project_key(db) or "",
|
|
parent_ticket=get_jira_parent_ticket(db) or "",
|
|
)
|
|
|
|
|
|
@router.patch("/jira-config", response_model=JiraConfigOut)
|
|
def update_jira_config(
|
|
payload: JiraConfigUpdate,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_role("admin")),
|
|
):
|
|
"""Update Jira configuration and persist to DB.
|
|
|
|
**Requires** the ``admin`` role. Only provided fields are updated.
|
|
"""
|
|
from app.services.jira_service import (
|
|
upsert_jira_config, get_jira_url, get_jira_project_key, is_jira_enabled, get_jira_parent_ticket,
|
|
)
|
|
|
|
update_data = payload.model_dump(exclude_unset=True)
|
|
for field, val in update_data.items():
|
|
db_key = _JIRA_KEYS.get(field)
|
|
if db_key:
|
|
upsert_jira_config(db, db_key, str(val))
|
|
db.commit()
|
|
|
|
return JiraConfigOut(
|
|
enabled=is_jira_enabled(db),
|
|
url=get_jira_url(db) or "",
|
|
project_key=get_jira_project_key(db) or "",
|
|
parent_ticket=get_jira_parent_ticket(db) or "",
|
|
)
|
|
|
|
|
|
@router.post("/jira-test")
|
|
def test_jira_connection(
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_role("admin")),
|
|
):
|
|
"""Test the Jira connection using the current user's credentials.
|
|
|
|
Requires the admin to have a personal Jira API token configured in their
|
|
profile settings.
|
|
"""
|
|
from app.services.jira_service import get_user_jira_client, get_jira_url
|
|
|
|
jira_url = get_jira_url(db)
|
|
if not jira_url:
|
|
raise HTTPException(status_code=400, detail="Jira URL not configured.")
|
|
|
|
try:
|
|
jira = get_user_jira_client(current_user, db)
|
|
# 10-second timeout so we never block Cloudflare into a 524
|
|
try:
|
|
jira._session.timeout = 10 # type: ignore[attr-defined]
|
|
except Exception:
|
|
pass
|
|
myself = jira.myself()
|
|
return {
|
|
"status": "ok",
|
|
"connected_as": myself.get("displayName") or myself.get("emailAddress", "unknown"),
|
|
"jira_url": jira_url,
|
|
}
|
|
except Exception as exc:
|
|
err = str(exc)
|
|
# Always return HTTP 200 with status="error" so Cloudflare never
|
|
# intercepts the response and the frontend always sees our message.
|
|
if "Expecting value" in err or "line 1 column 1" in err:
|
|
msg = (
|
|
"Jira returned a non-JSON response. "
|
|
"Verify the URL (e.g. https://company.atlassian.net), "
|
|
"email and API token."
|
|
)
|
|
elif "401" in err or "Unauthorized" in err:
|
|
msg = "Authentication failed (401). Check your Atlassian email and API token."
|
|
elif "403" in err or "Forbidden" in err:
|
|
msg = "Access denied (403). The token may not have permission for this Jira project."
|
|
elif "timed out" in err.lower() or "timeout" in err.lower():
|
|
msg = "Connection timed out. Check that the Jira URL is reachable from the server."
|
|
elif "not configured" in err.lower():
|
|
msg = err
|
|
else:
|
|
msg = f"Jira connection failed: {err}"
|
|
logger.warning("Jira test connection failed: %s", err)
|
|
return {"status": "error", "message": msg, "jira_url": jira_url}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /system/email-config
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("/email-config", response_model=EmailConfigOut)
|
|
def get_email_config(
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_role("admin")),
|
|
):
|
|
"""Return current SMTP email configuration (merged DB + env).
|
|
|
|
**Requires** the ``admin`` role. Password is never returned.
|
|
"""
|
|
cfg = _read_email_config_from_db(db)
|
|
return EmailConfigOut(
|
|
enabled=cfg["enabled"],
|
|
host=cfg["host"],
|
|
port=cfg["port"],
|
|
username=cfg["username"],
|
|
from_email=cfg["from_email"],
|
|
use_tls=cfg["use_tls"],
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# PATCH /system/email-config
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.patch("/email-config", response_model=EmailConfigOut)
|
|
def update_email_config(
|
|
payload: EmailConfigUpdate,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_role("admin")),
|
|
):
|
|
"""Update SMTP email configuration and persist to DB.
|
|
|
|
**Requires** the ``admin`` role.
|
|
Only provided fields are updated (partial update).
|
|
"""
|
|
update_data = payload.model_dump(exclude_unset=True)
|
|
for field, val in update_data.items():
|
|
db_key = _SMTP_KEYS.get(field)
|
|
if db_key:
|
|
_upsert_config(db, db_key, str(val))
|
|
db.commit()
|
|
|
|
cfg = _read_email_config_from_db(db)
|
|
return EmailConfigOut(
|
|
enabled=cfg["enabled"],
|
|
host=cfg["host"],
|
|
port=cfg["port"],
|
|
username=cfg["username"],
|
|
from_email=cfg["from_email"],
|
|
use_tls=cfg["use_tls"],
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /system/email-test
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.post("/email-test")
|
|
def send_test_email(
|
|
payload: EmailTestRequest,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_role("admin")),
|
|
):
|
|
"""Send a test email to verify SMTP configuration.
|
|
|
|
**Requires** the ``admin`` role.
|
|
Returns 200 on success, 502 if sending fails.
|
|
"""
|
|
from app.services.email_service import send_test_email as _send_test
|
|
|
|
ok = _send_test(payload.to, db=db)
|
|
if not ok:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_502_BAD_GATEWAY,
|
|
detail="Failed to send test email. Check SMTP configuration and server logs.",
|
|
)
|
|
return {"detail": f"Test email sent to {payload.to}"}
|