"""Generic outbound-email sender — POSTs to an admin-configured webhook. Replaces the old SMTP-based email system. Every notification email in the platform (password setup/reset, test validated, campaign completed, new MITRE techniques synced, etc.) goes through ``send_webhook_email`` here, which POSTs a ``{to, subject, body}`` JSON payload to a single configured webhook URL — intended for a Power Automate flow that actually delivers the email (the ``body`` is full HTML; the flow's "send email" step must have its "Is HTML" option enabled). An optional API key, if configured, is sent as an ``x-api-key`` header. SMTP (``app.services.email_service``) is intentionally left in place but unused and unreachable from the UI — see that module's docstring. """ from __future__ import annotations import base64 import html as html_lib import logging import re from pathlib import Path import requests from sqlalchemy.orm import Session logger = logging.getLogger(__name__) _WEBHOOK_URL_CONFIG_KEY = "email_webhook.url" _WEBHOOK_API_KEY_CONFIG_KEY = "email_webhook.api_key" _LOGO_PATH = Path(__file__).resolve().parent.parent / "assets" / "email_logo.png" _URL_RE = re.compile(r"^https?://\S+$") _logo_b64_cache: str | None = None def _get_logo_base64() -> str: """Read + cache the inline email logo as a base64 string (empty if missing).""" global _logo_b64_cache if _logo_b64_cache is None: try: _logo_b64_cache = base64.b64encode(_LOGO_PATH.read_bytes()).decode("ascii") except OSError: logger.warning("Email logo asset missing at %s", _LOGO_PATH) _logo_b64_cache = "" return _logo_b64_cache def _message_to_html(message: str) -> str: """Turn a plain-text message (paragraphs separated by blank lines) into escaped HTML, rendering any lone-URL paragraph as a call-to-action button. """ parts: list[str] = [] for para in message.strip().split("\n\n"): para = para.strip() if not para: continue if _URL_RE.match(para): url = html_lib.escape(para, quote=True) parts.append( f'

' f'Open Link →' f'

' ) else: escaped = html_lib.escape(para).replace("\n", "
") parts.append(f'

{escaped}

') return "".join(parts) def build_email_body(full_name: str | None, message: str) -> str: """Wrap *message* in Aegis's branded HTML email template (inline logo).""" greeting = html_lib.escape(full_name) if full_name else "there" message_html = _message_to_html(message) logo_b64 = _get_logo_base64() logo_tag = ( f'Aegis' if logo_b64 else "" ) return f"""
{logo_tag}
AEGIS SECURITY PLATFORM
Purple Team Engineering

Hi {greeting},

{message_html}

Assume breach. Validate controls. Improve continuously.

Owned and operated by Enterprise Corp. This is an automated notification — please do not reply.
""" def _read_system_config(db: Session, key: str) -> str | None: from app.models.system_config import SystemConfig # avoid circular at import time row = db.query(SystemConfig).filter(SystemConfig.key == key).first() return row.value if row else None def _write_system_config(db: Session, key: str, value: str) -> None: from app.models.system_config import SystemConfig row = db.query(SystemConfig).filter(SystemConfig.key == key).first() if row: row.value = value else: db.add(SystemConfig(key=key, value=value)) def get_webhook_url(db: Session) -> str | None: """Return the configured Power-Automate-style webhook URL, or None.""" return _read_system_config(db, _WEBHOOK_URL_CONFIG_KEY) or None def set_webhook_url(db: Session, url: str) -> None: """Persist the webhook URL. Does not commit; caller commits.""" _write_system_config(db, _WEBHOOK_URL_CONFIG_KEY, url) def get_webhook_api_key(db: Session) -> str | None: """Return the configured webhook API key, or None.""" return _read_system_config(db, _WEBHOOK_API_KEY_CONFIG_KEY) or None def set_webhook_api_key(db: Session, api_key: str) -> None: """Persist the webhook API key. Does not commit; caller commits.""" _write_system_config(db, _WEBHOOK_API_KEY_CONFIG_KEY, api_key) def send_webhook_email( db: Session, *, to: str | None, subject: str, message: str, full_name: str | None = None, ) -> bool: """POST a notification email to the configured webhook. Best-effort: returns False (and logs a warning) if no webhook is configured, *to* is empty, or the request fails — never raises, since a webhook outage must not break whatever triggered the notification. """ if not to: return False webhook_url = get_webhook_url(db) if not webhook_url: logger.warning("Email webhook is not configured — dropping email to %s", to) return False api_key = get_webhook_api_key(db) headers = {"x-api-key": api_key} if api_key else {} try: requests.post( webhook_url, json={"to": to, "subject": subject, "body": build_email_body(full_name, message)}, headers=headers, timeout=10, ) return True except requests.RequestException: logger.warning("Email webhook call failed for %s", to, exc_info=True) return False