kitos/Aegis
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity
Files
c99cc4946ae6dd49a024bb63145ccd64e41bbff3
Aegis/backend/app/routers/test_templates.py
T
kitos c99cc4946a refactor(docs+comments): add Google-style docstrings and inline comments across backend 
Task D — Google-style docstrings (Args/Returns) on every public function,
method, and class across all 158 Python files in the backend. Zero ruff D
violations (pydocstyle Google convention).

Task E — Explanatory one-line comment before every code line (~11600 new
comments). ruff check passes clean after isort re-sort.
2026-06-10 13:25:14 +02:00

524 lines
18 KiB
Python
Raw Blame History

"""CRUD router for TestTemplates — predefined test catalog.
Endpoints
---------
GET /test-templates — list with filters + pagination
GET /test-templates/stats — catalog statistics (admin)
GET /test-templates/{id} — detail
POST /test-templates — create custom (admin)
PATCH /test-templates/{id} — update (admin)
PATCH /test-templates/{id}/toggle-active — toggle active/inactive (admin)
DELETE /test-templates/{id} — soft delete (admin)
GET /test-templates/by-technique/{mitre_id} — templates for a MITRE technique
Filters (GET /test-templates)
-----------------------------
- source: atomic_red_team | mitre | custom
- platform: windows | linux | macos
- severity: low | medium | high | critical
- mitre_technique_id: filter by specific technique
- search: full-text search across name and description
- is_active: true | false (default only active)
- offset / limit: pagination (default limit=50)
"""
# Import uuid
import uuid
# Import Optional from typing
from typing import Optional
# Import APIRouter, Depends, Query, status from fastapi
from fastapi import APIRouter, Depends, Query, status
# Import Session from sqlalchemy.orm
from sqlalchemy.orm import Session
# Import get_db from app.database
from app.database import get_db
# Import get_current_user, require_any_role from app.dependencies.auth
from app.dependencies.auth import get_current_user, require_any_role
# Import UnitOfWork from app.domain.unit_of_work
from app.domain.unit_of_work import UnitOfWork
# Import User from app.models.user
from app.models.user import User
# Import from app.schemas.test_template
from app.schemas.test_template import (
TestTemplateCreate,
TestTemplateOut,
TestTemplateSummary,
)
# Import log_action from app.services.audit_service
from app.services.audit_service import log_action
# Import from app.services.test_template_service
from app.services.test_template_service import (
bulk_activate,
get_template_or_raise,
get_template_stats,
list_templates,
soft_delete_template,
)
# Import from app.services.test_template_service
from app.services.test_template_service import (
create_template as create_template_svc,
)
# Import from app.services.test_template_service
from app.services.test_template_service import (
get_templates_by_technique as templates_by_technique,
)
# Import from app.services.test_template_service
from app.services.test_template_service import (
toggle_template_active as toggle_template_active_svc,
)
# Import from app.services.test_template_service
from app.services.test_template_service import (
update_template as update_template_svc,
)
# Assign router = APIRouter(prefix="/test-templates", tags=["test-templates"])
router = APIRouter(prefix="/test-templates", tags=["test-templates"])
# ---------------------------------------------------------------------------
# GET /test-templates — list with filters + pagination
# ---------------------------------------------------------------------------
@router.get("", response_model=list[TestTemplateSummary])
# Define function _list_templates_handler
def _list_templates_handler(
# Entry: source
source: Optional[str] = Query(None, description="Filter by source (atomic_red_team, mitre, custom)"),
# Entry: platform
platform: Optional[str] = Query(None, description="Filter by platform (windows, linux, macos)"),
# Entry: severity
severity: Optional[str] = Query(None, description="Filter by severity (low, medium, high, critical)"),
# Entry: mitre_technique_id
mitre_technique_id: Optional[str] = Query(None, description="Filter by MITRE technique ID"),
# Entry: search
search: Optional[str] = Query(None, description="Search in name and description"),
# Entry: is_active
is_active: Optional[bool] = Query(None, description="Filter by active status (true/false). Omit to return all."),
# Entry: offset
offset: int = Query(0, ge=0),
# Entry: limit
limit: int = Query(50, ge=1, le=200),
# Entry: db
db: Session = Depends(get_db),
# Entry: current_user
current_user: User = Depends(get_current_user),
) -> list:
"""Return a paginated, filterable list of test templates.
Args:
source (Optional[str]): Filter by source (``atomic_red_team``, ``mitre``, ``custom``).
platform (Optional[str]): Filter by platform (``windows``, ``linux``, ``macos``).
severity (Optional[str]): Filter by severity (``low``, ``medium``, ``high``, ``critical``).
mitre_technique_id (Optional[str]): Filter by MITRE technique ID string.
search (Optional[str]): Full-text search across name and description.
is_active (Optional[bool]): Filter by active status; omit to return all.
offset (int): Number of records to skip for pagination.
limit (int): Maximum number of records to return.
db (Session): SQLAlchemy database session.
current_user (User): Authenticated user making the request.
Returns:
list: Serialised list of :class:`TestTemplateSummary` objects.
"""
# Return list_templates(
return list_templates(
db,
# Keyword argument: source
source=source,
# Keyword argument: platform
platform=platform,
# Keyword argument: severity
severity=severity,
# Keyword argument: mitre_technique_id
mitre_technique_id=mitre_technique_id,
# Keyword argument: search
search=search,
# Keyword argument: is_active
is_active=is_active,
# Keyword argument: offset
offset=offset,
# Keyword argument: limit
limit=limit,
)
# ---------------------------------------------------------------------------
# GET /test-templates/stats — catalog statistics (admin)
# ---------------------------------------------------------------------------
@router.get("/stats")
# Define function template_stats
def template_stats(
# Entry: db
db: Session = Depends(get_db),
# Entry: current_user
current_user: User = Depends(require_any_role("red_lead", "blue_lead")),
) -> dict:
"""Return catalog statistics: active, by_source, by_platform.
Args:
db (Session): SQLAlchemy database session.
current_user (User): Authenticated red_lead or blue_lead.
Returns:
dict: Counts of active templates broken down by source and platform.
"""
# Return get_template_stats(db)
return get_template_stats(db)
# ---------------------------------------------------------------------------
# PATCH /test-templates/bulk-activate — activate/deactivate all (admin)
# ---------------------------------------------------------------------------
@router.patch("/bulk-activate")
# Define function bulk_activate_templates
def bulk_activate_templates(
# Entry: activate
activate: bool = Query(True, description="True to activate all, False to deactivate all"),
# Entry: db
db: Session = Depends(get_db),
# Entry: current_user
current_user: User = Depends(require_any_role("red_lead", "blue_lead")),
) -> dict:
"""Set all templates to active or inactive.
Args:
activate (bool): ``True`` to activate all templates, ``False`` to deactivate all.
db (Session): SQLAlchemy database session.
current_user (User): Authenticated red_lead or blue_lead.
Returns:
dict: Confirmation message with ``affected`` count and the applied ``is_active`` flag.
"""
# Assign count = bulk_activate(db, activate=activate)
count = bulk_activate(db, activate=activate)
# Open context manager
with UnitOfWork(db) as uow:
# Call log_action()
log_action(
db,
# Keyword argument: user_id
user_id=current_user.id,
# Keyword argument: action
action="bulk_activate_templates" if activate else "bulk_deactivate_templates",
# Keyword argument: entity_type
entity_type="test_template",
# Keyword argument: entity_id
entity_id=None,
# Keyword argument: details
details={"affected": count, "is_active": activate},
)
# Call uow.commit()
uow.commit()
# Return {
return {
# Literal argument value
"detail": f"{'Activated' if activate else 'Deactivated'} {count} templates",
# Literal argument value
"affected": count,
# Literal argument value
"is_active": activate,
}
# ---------------------------------------------------------------------------
# GET /test-templates/by-technique/{mitre_id}
# ---------------------------------------------------------------------------
@router.get("/by-technique/{mitre_id}", response_model=list[TestTemplateSummary])
# Define function _templates_by_technique_handler
def _templates_by_technique_handler(
# Entry: mitre_id
mitre_id: str,
# Entry: db
db: Session = Depends(get_db),
# Entry: current_user
current_user: User = Depends(get_current_user),
) -> list:
"""Return all active templates mapped to a specific MITRE technique.
Args:
mitre_id (str): MITRE ATT&CK technique ID (e.g. ``T1059.001``).
db (Session): SQLAlchemy database session.
current_user (User): Authenticated user making the request.
Returns:
list: Serialised list of :class:`TestTemplateSummary` objects for the technique.
"""
# Return templates_by_technique(db, mitre_id)
return templates_by_technique(db, mitre_id)
# ---------------------------------------------------------------------------
# GET /test-templates/{id} — detail
# ---------------------------------------------------------------------------
@router.get("/{template_id}", response_model=TestTemplateOut)
# Define function get_template
def get_template(
# Entry: template_id
template_id: uuid.UUID,
# Entry: db
db: Session = Depends(get_db),
# Entry: current_user
current_user: User = Depends(get_current_user),
) -> TestTemplateOut:
"""Return full details for a single test template.
Args:
template_id (uuid.UUID): Primary key of the template to retrieve.
db (Session): SQLAlchemy database session.
current_user (User): Authenticated user making the request.
Returns:
TestTemplateOut: Full template detail including all fields.
"""
# Return get_template_or_raise(db, template_id)
return get_template_or_raise(db, template_id)
# ---------------------------------------------------------------------------
# POST /test-templates — create (admin only)
# ---------------------------------------------------------------------------
@router.post(
# Literal argument value
"",
# Keyword argument: response_model
response_model=TestTemplateOut,
# Keyword argument: status_code
status_code=status.HTTP_201_CREATED,
)
# Define function create_template
def create_template(
# Entry: payload
payload: TestTemplateCreate,
# Entry: db
db: Session = Depends(get_db),
# Entry: current_user
current_user: User = Depends(require_any_role("red_lead", "blue_lead")),
) -> TestTemplateOut:
"""Create a custom test template.
Args:
payload (TestTemplateCreate): All fields for the new template.
db (Session): SQLAlchemy database session.
current_user (User): Authenticated red_lead or blue_lead creating the template.
Returns:
TestTemplateOut: The newly created template with all fields populated.
"""
# Assign template = create_template_svc(db, **payload.model_dump())
template = create_template_svc(db, **payload.model_dump())
# Open context manager
with UnitOfWork(db) as uow:
# Call log_action()
log_action(
db,
# Keyword argument: user_id
user_id=current_user.id,
# Keyword argument: action
action="create_test_template",
# Keyword argument: entity_type
entity_type="test_template",
# Keyword argument: entity_id
entity_id=template.id,
# Keyword argument: details
details={
# Literal argument value
"name": template.name,
# Literal argument value
"source": template.source,
# Literal argument value
"mitre_technique_id": template.mitre_technique_id,
},
)
# Call uow.commit()
uow.commit()
# Reload ORM object attributes from the database
db.refresh(template)
# Return template
return template
# ---------------------------------------------------------------------------
# PATCH /test-templates/{id} — update (admin only)
# ---------------------------------------------------------------------------
@router.patch("/{template_id}", response_model=TestTemplateOut)
# Define function update_template
def update_template(
# Entry: template_id
template_id: uuid.UUID,
# Entry: payload
payload: TestTemplateCreate,
# Entry: db
db: Session = Depends(get_db),
# Entry: current_user
current_user: User = Depends(require_any_role("red_lead", "blue_lead")),
) -> TestTemplateOut:
"""Update fields of an existing test template.
Args:
template_id (uuid.UUID): Primary key of the template to update.
payload (TestTemplateCreate): Fields to update; only set fields are applied.
db (Session): SQLAlchemy database session.
current_user (User): Authenticated red_lead or blue_lead updating the template.
Returns:
TestTemplateOut: The updated template with refreshed field values.
"""
# Assign template = update_template_svc(db, template_id, **payload.model_dump(exclude_u...
template = update_template_svc(db, template_id, **payload.model_dump(exclude_unset=True))
# Open context manager
with UnitOfWork(db) as uow:
# Call log_action()
log_action(
db,
# Keyword argument: user_id
user_id=current_user.id,
# Keyword argument: action
action="update_test_template",
# Keyword argument: entity_type
entity_type="test_template",
# Keyword argument: entity_id
entity_id=template.id,
# Keyword argument: details
details={"updated_fields": list(payload.model_dump(exclude_unset=True).keys())},
)
# Call uow.commit()
uow.commit()
# Reload ORM object attributes from the database
db.refresh(template)
# Return template
return template
# ---------------------------------------------------------------------------
# PATCH /test-templates/{id}/toggle-active — toggle active/inactive (admin)
# ---------------------------------------------------------------------------
@router.patch("/{template_id}/toggle-active", response_model=TestTemplateOut)
# Define function toggle_template_active
def toggle_template_active(
# Entry: template_id
template_id: uuid.UUID,
# Entry: db
db: Session = Depends(get_db),
# Entry: current_user
current_user: User = Depends(require_any_role("red_lead", "blue_lead")),
) -> TestTemplateOut:
"""Toggle a template between active and inactive (is_active = not is_active).
Args:
template_id (uuid.UUID): Primary key of the template to toggle.
db (Session): SQLAlchemy database session.
current_user (User): Authenticated red_lead or blue_lead.
Returns:
TestTemplateOut: The template with the updated ``is_active`` flag.
"""
# Assign template = toggle_template_active_svc(db, template_id)
template = toggle_template_active_svc(db, template_id)
# Open context manager
with UnitOfWork(db) as uow:
# Call log_action()
log_action(
db,
# Keyword argument: user_id
user_id=current_user.id,
# Keyword argument: action
action="toggle_test_template",
# Keyword argument: entity_type
entity_type="test_template",
# Keyword argument: entity_id
entity_id=template.id,
# Keyword argument: details
details={"name": template.name, "is_active": template.is_active},
)
# Call uow.commit()
uow.commit()
# Reload ORM object attributes from the database
db.refresh(template)
# Return template
return template
# ---------------------------------------------------------------------------
# DELETE /test-templates/{id} — soft delete (admin only)
# ---------------------------------------------------------------------------
@router.delete("/{template_id}", status_code=status.HTTP_200_OK)
# Define function delete_template
def delete_template(
# Entry: template_id
template_id: uuid.UUID,
# Entry: db
db: Session = Depends(get_db),
# Entry: current_user
current_user: User = Depends(require_any_role("red_lead", "blue_lead")),
) -> dict:
"""Soft-delete a test template by setting ``is_active=False``.
Args:
template_id (uuid.UUID): Primary key of the template to delete.
db (Session): SQLAlchemy database session.
current_user (User): Authenticated red_lead or blue_lead.
Returns:
dict: Confirmation message with key ``detail``.
"""
# Assign template = get_template_or_raise(db, template_id)
template = get_template_or_raise(db, template_id)
# Call soft_delete_template()
soft_delete_template(db, template_id)
# Open context manager
with UnitOfWork(db) as uow:
# Call log_action()
log_action(
db,
# Keyword argument: user_id
user_id=current_user.id,
# Keyword argument: action
action="delete_test_template",
# Keyword argument: entity_type
entity_type="test_template",
# Keyword argument: entity_id
entity_id=template.id,
# Keyword argument: details
details={"name": template.name},
)
# Call uow.commit()
uow.commit()
# Return {"detail": "Test template deactivated"}
return {"detail": "Test template deactivated"}
Reference in New Issue View Git Blame Copy Permalink