c1e06d4c0a
Some checks failed
Aegis CI / lint-and-test (push) Has been cancelled
- Phase 6.1: WebhookConfig model, CRUD router (/api/v1/webhooks, admin-only), dispatch_webhook() with HMAC signing; integrated into test validation, campaign completion, and MITRE sync job - Phase 7.1: SMTP email service with send_test_validated_email, send_campaign_completed_email, send_new_mitre_techniques_email; notify_role_with_email() added to notification_service - Phase 7.2: notification_preferences and jira_account_id on User model; PATCH /users/me/preferences endpoint; Alembic migrations b031phase6 and b032phase7 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
150 lines
4.9 KiB
Python
150 lines
4.9 KiB
Python
"""Webhook configuration CRUD router — admin only.
|
|
|
|
Endpoints
|
|
---------
|
|
GET /webhooks — list all webhook configs
|
|
POST /webhooks — create a new webhook config
|
|
GET /webhooks/{id} — get a single webhook config
|
|
PATCH /webhooks/{id} — update a webhook config
|
|
DELETE /webhooks/{id} — hard-delete a webhook config
|
|
POST /webhooks/{id}/test — send a test ping
|
|
"""
|
|
|
|
import uuid
|
|
|
|
from fastapi import APIRouter, Depends, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.database import get_db
|
|
from app.dependencies.auth import require_role
|
|
from app.domain.unit_of_work import UnitOfWork
|
|
from app.models.user import User
|
|
from app.schemas.webhook import WebhookConfigCreate, WebhookConfigOut, WebhookConfigUpdate
|
|
from app.services.webhook_service import (
|
|
create_webhook,
|
|
delete_webhook,
|
|
dispatch_webhook,
|
|
get_webhook_or_raise,
|
|
list_webhooks,
|
|
update_webhook,
|
|
)
|
|
|
|
router = APIRouter(prefix="/webhooks", tags=["webhooks"])
|
|
|
|
|
|
def _mask_secret(wh) -> WebhookConfigOut:
|
|
"""Return a WebhookConfigOut with the secret masked."""
|
|
out = WebhookConfigOut.model_validate(wh)
|
|
if wh.secret:
|
|
out.secret = "***"
|
|
else:
|
|
out.secret = None
|
|
return out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /webhooks
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("", response_model=list[WebhookConfigOut])
|
|
def list_webhooks_route(
|
|
offset: int = 0,
|
|
limit: int = 50,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_role("admin")),
|
|
):
|
|
"""Return all webhook configurations. **Requires admin role.**"""
|
|
webhooks = list_webhooks(db, offset=offset, limit=limit)
|
|
return [_mask_secret(wh) for wh in webhooks]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /webhooks
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.post("", response_model=WebhookConfigOut, status_code=status.HTTP_201_CREATED)
|
|
def create_webhook_route(
|
|
payload: WebhookConfigCreate,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_role("admin")),
|
|
):
|
|
"""Create a new webhook configuration. **Requires admin role.**"""
|
|
with UnitOfWork(db) as uow:
|
|
wh = create_webhook(db, created_by=current_user.id, payload=payload)
|
|
uow.commit()
|
|
db.refresh(wh)
|
|
return _mask_secret(wh)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /webhooks/{id}
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("/{webhook_id}", response_model=WebhookConfigOut)
|
|
def get_webhook_route(
|
|
webhook_id: uuid.UUID,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_role("admin")),
|
|
):
|
|
"""Return a single webhook configuration. **Requires admin role.**"""
|
|
wh = get_webhook_or_raise(db, webhook_id)
|
|
return _mask_secret(wh)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# PATCH /webhooks/{id}
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.patch("/{webhook_id}", response_model=WebhookConfigOut)
|
|
def update_webhook_route(
|
|
webhook_id: uuid.UUID,
|
|
payload: WebhookConfigUpdate,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_role("admin")),
|
|
):
|
|
"""Update one or more fields of a webhook configuration. **Requires admin role.**"""
|
|
with UnitOfWork(db) as uow:
|
|
wh = update_webhook(db, webhook_id, payload)
|
|
uow.commit()
|
|
db.refresh(wh)
|
|
return _mask_secret(wh)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DELETE /webhooks/{id}
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.delete("/{webhook_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
def delete_webhook_route(
|
|
webhook_id: uuid.UUID,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_role("admin")),
|
|
):
|
|
"""Hard-delete a webhook configuration. **Requires admin role.**"""
|
|
with UnitOfWork(db) as uow:
|
|
delete_webhook(db, webhook_id)
|
|
uow.commit()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /webhooks/{id}/test
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.post("/{webhook_id}/test", status_code=status.HTTP_202_ACCEPTED)
|
|
def test_webhook_route(
|
|
webhook_id: uuid.UUID,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_role("admin")),
|
|
):
|
|
"""Send a test ping to the webhook endpoint. **Requires admin role.**"""
|
|
# Verify the webhook exists before dispatching
|
|
get_webhook_or_raise(db, webhook_id)
|
|
dispatch_webhook("webhook.test", {"webhook_id": str(webhook_id), "message": "Test ping from Aegis"})
|
|
return {"detail": "Test ping dispatched"}
|