feat: Add email contact annotation

Author: dadav

backend/alembic/versions/005_add_escalation_warning_days.py

@@ -0,0 +1,24 @@
+"""Add escalation_warning_days to global_settings
+
+Revision ID: 005
+Revises: 004
+Create Date: 2026-03-02 00:00:00.000000
+"""
+from alembic import op
+import sqlalchemy as sa
+
+revision = "005"
+down_revision = "004"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "global_settings",
+        sa.Column("escalation_warning_days", sa.Integer(), nullable=False, server_default="3"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("global_settings", "escalation_warning_days")

backend/alembic/versions/006_add_namespace_contacts.py

@@ -0,0 +1,29 @@
+"""Add namespace_contacts table for escalation emails
+
+Revision ID: 006
+Revises: 005
+Create Date: 2026-03-02 00:00:00.000000
+"""
+from alembic import op
+import sqlalchemy as sa
+
+revision = "006"
+down_revision = "005"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "namespace_contacts",
+        sa.Column("id", sa.Uuid(), primary_key=True),
+        sa.Column("namespace", sa.String(255), nullable=False, index=True),
+        sa.Column("cluster_name", sa.String(255), nullable=False),
+        sa.Column("escalation_email", sa.String(255), nullable=False),
+        sa.Column("updated_at", sa.DateTime(), nullable=True),
+        sa.UniqueConstraint("namespace", "cluster_name", name="uq_namespace_contact_ns_cluster"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("namespace_contacts")

backend/app/auth/middleware.py

@@ -7,6 +7,7 @@
 
 from ..config import settings
 from ..database import AppSessionLocal
+from ..models.namespace_contact import NamespaceContact
 from ..models.user import User, UserRole
 
 logger = logging.getLogger(__name__)
@@ -28,6 +29,50 @@ def _parse_namespaces_header(raw: str) -> list[tuple[str, str]]:
     return pairs
 
 
+def _parse_namespace_emails_header(raw: str) -> list[tuple[str, str, str]]:
+    """Parse 'ns1:cluster1=email@x.com,ns2:cluster2=email@y.com' into [(ns, cluster, email), ...]."""
+    if not raw.strip():
+        return []
+    result = []
+    for entry in raw.split(","):
+        entry = entry.strip()
+        if "=" not in entry:
+            continue
+        ns_cluster, email = entry.rsplit("=", 1)
+        email = email.strip()
+        if ":" not in ns_cluster or not email:
+            continue
+        ns, cluster = ns_cluster.split(":", 1)
+        ns, cluster = ns.strip(), cluster.strip()
+        if ns and cluster:
+            result.append((ns, cluster, email))
+    return result
+
+
+async def _upsert_namespace_contacts(
+    session: AsyncSession, contacts: list[tuple[str, str, str]]
+) -> None:
+    """Upsert namespace escalation email contacts. Only writes if data changed."""
+    if not contacts:
+        return
+    for ns, cluster, email in contacts:
+        result = await session.execute(
+            select(NamespaceContact).where(
+                NamespaceContact.namespace == ns,
+                NamespaceContact.cluster_name == cluster,
+            )
+        )
+        existing = result.scalar_one_or_none()
+        if existing:
+            if existing.escalation_email != email:
+                existing.escalation_email = email
+        else:
+            session.add(NamespaceContact(
+                namespace=ns, cluster_name=cluster, escalation_email=email,
+            ))
+    await session.commit()
+
+
 class CurrentUser:
     def __init__(self, id: str, username: str, email: str, role: UserRole, namespaces: list[tuple[str, str]]):
         self.id = id
@@ -94,6 +139,11 @@ def _to_current_user(user: User, namespaces: list[tuple[str, str]]) -> CurrentUs
 async def _handle_dev_mode(session: AsyncSession) -> CurrentUser:
     namespaces = _parse_namespaces_header(settings.dev_user_namespaces)
 
+    # Upsert dev namespace email contacts
+    ns_emails = _parse_namespace_emails_header(settings.dev_namespace_emails)
+    if ns_emails:
+        await _upsert_namespace_contacts(session, ns_emails)
+
     user_data = {
         "id": settings.dev_user_id,
         "username": settings.dev_user_name,
@@ -133,6 +183,12 @@ async def _handle_spoke_proxy(session: AsyncSession, request: Request) -> Curren
     user = await _get_or_create_user(session, user_data)
     user = await _sync_user_fields(session, user, user_data)
 
+    # Upsert namespace escalation email contacts from header
+    ns_emails_raw = request.headers.get("X-Forwarded-Namespace-Emails", "")
+    ns_emails = _parse_namespace_emails_header(ns_emails_raw)
+    if ns_emails:
+        await _upsert_namespace_contacts(session, ns_emails)
+
     logger.info("Spoke proxy auth: user=%s, role=%s, namespaces=%d", user_id, role.value, len(namespaces))
     return _to_current_user(user, namespaces)
 

backend/app/config.py

@@ -24,6 +24,7 @@ class Settings(BaseSettings):
     dev_user_email: str = Field(default="dev@example.com")
     dev_user_role: str = Field(default="sec_team")  # "sec_team" or "team_member"
     dev_user_namespaces: str = Field(default="")  # format: ns1:cluster1,ns2:cluster2
+    dev_namespace_emails: str = Field(default="")  # format: ns1:cluster1=email@example.com,ns2:cluster2=other@example.com
 
     # OIDC (production)
     oidc_issuer: str = Field(default="")

backend/app/mail/service.py

@@ -80,6 +80,26 @@ async def send_risk_status_email(
     )
 
 
+async def send_escalation_email(
+    to_email: str,
+    cve_id: str,
+    namespace: str,
+    cluster_name: str,
+    level: int,
+    base_url: str | None = None,
+) -> None:
+    base_url = base_url or settings.app_base_url
+    tmpl = _jinja_env.get_template("escalation.html")
+    html = tmpl.render(
+        cve_id=cve_id,
+        namespace=namespace,
+        cluster_name=cluster_name,
+        level=level,
+        link=f"{base_url}/eskalationen",
+    )
+    await send_email(to_email, f"CVE-Eskalation Stufe {level}: {cve_id}", html)
+
+
 async def send_weekly_digest(
     to_email: str,
     stats: dict,

backend/app/mail/templates/escalation.html

@@ -0,0 +1,34 @@
+<!DOCTYPE html>
+<html lang="de">
+<head><meta charset="utf-8"><title>Eskalation: {{ cve_id }}</title></head>
+<body style="font-family:Arial,sans-serif;background:#f5f5f5;margin:0;padding:20px">
+<div style="max-width:600px;margin:auto;background:#fff;border-radius:4px;overflow:hidden;box-shadow:0 1px 4px rgba(0,0,0,.1)">
+  <div style="background:#151515;padding:20px 30px">
+    <h1 style="color:#fff;margin:0;font-size:18px">RHACS CVE Manager</h1>
+  </div>
+  <div style="padding:30px">
+    <h2 style="color:#c9190b;margin-top:0">Eskalation Stufe {{ level }}</h2>
+    <p>Die folgende CVE wurde auf <strong>Stufe {{ level }}</strong> eskaliert:</p>
+    <table style="width:100%;border-collapse:collapse;margin:16px 0">
+      <tr style="background:#f0f0f0">
+        <td style="padding:8px 12px;font-weight:bold">CVE</td>
+        <td style="padding:8px 12px">{{ cve_id }}</td>
+      </tr>
+      <tr>
+        <td style="padding:8px 12px;font-weight:bold">Namespace</td>
+        <td style="padding:8px 12px">{{ namespace }}</td>
+      </tr>
+      <tr style="background:#f0f0f0">
+        <td style="padding:8px 12px;font-weight:bold">Cluster</td>
+        <td style="padding:8px 12px">{{ cluster_name }}</td>
+      </tr>
+    </table>
+    <p>Bitte prüfen Sie die CVE und ergreifen Sie entsprechende Maßnahmen.</p>
+    <a href="{{ link }}" style="display:inline-block;background:#0066cc;color:#fff;padding:10px 20px;text-decoration:none;border-radius:4px;margin-top:10px">Eskalation ansehen</a>
+  </div>
+  <div style="background:#f0f0f0;padding:16px 30px;font-size:12px;color:#666">
+    Dies ist eine automatische Benachrichtigung vom RHACS CVE Manager.
+  </div>
+</div>
+</body>
+</html>

backend/app/models/__init__.py

@@ -4,6 +4,7 @@
 from .cve_priority import CvePriority, PriorityLevel
 from .escalation import Escalation
 from .global_settings import GlobalSettings
+from .namespace_contact import NamespaceContact
 from .notification import Notification, NotificationType
 from .risk_acceptance import RiskAcceptance, RiskAcceptanceComment, RiskStatus
 from .user import User, UserRole
@@ -23,4 +24,5 @@
     "Notification",
     "NotificationType",
     "AuditLog",
+    "NamespaceContact",
 ]

backend/app/models/global_settings.py

@@ -41,6 +41,7 @@ class GlobalSettings(Base):
     escalation_rules: Mapped[list] = mapped_column(
         JSONB, nullable=False, default=lambda: DEFAULT_ESCALATION_RULES
     )
+    escalation_warning_days: Mapped[int] = mapped_column(Integer, nullable=False, default=3)
     digest_day: Mapped[int] = mapped_column(Integer, nullable=False, default=0)  # 0=Monday
     management_email: Mapped[str] = mapped_column(String(255), nullable=False, default="")
     updated_by: Mapped[str | None] = mapped_column(

backend/app/models/namespace_contact.py

@@ -0,0 +1,20 @@
+from datetime import datetime
+from uuid import UUID, uuid4
+
+from sqlalchemy import String, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from ..database import Base
+
+
+class NamespaceContact(Base):
+    __tablename__ = "namespace_contacts"
+    __table_args__ = (
+        UniqueConstraint("namespace", "cluster_name", name="uq_namespace_contact_ns_cluster"),
+    )
+
+    id: Mapped[UUID] = mapped_column(primary_key=True, default=uuid4)
+    namespace: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
+    cluster_name: Mapped[str] = mapped_column(String(255), nullable=False)
+    escalation_email: Mapped[str] = mapped_column(String(255), nullable=False)
+    updated_at: Mapped[datetime] = mapped_column(default=datetime.utcnow, onupdate=datetime.utcnow)

backend/app/routers/dashboard.py

@@ -22,6 +22,7 @@
     ThresholdPreview,
 )
 from ..schemas.cve import CveListItem, SeverityLevel
+from ..services.escalation_preview import compute_upcoming_escalations
 from ..stackrox import queries as sx
 
 router = APIRouter(prefix="/dashboard", tags=["dashboard"])
@@ -88,6 +89,7 @@ async def team_dashboard(
             return TeamDashboardData(
                 stat_total_cves=0,
                 stat_escalations=0,
+                stat_upcoming_escalations=0,
                 stat_fixable_critical_cves=0,
                 stat_open_risk_acceptances=0,
                 severity_distribution=[], cves_per_namespace=[], priority_cves=[],
@@ -168,6 +170,12 @@ async def team_dashboard(
     )
     trend = await sx.get_cve_trend(sx_db, ns_list_for_queries)
 
+    # Upcoming escalation count
+    upcoming_escalations = []
+    if settings:
+        upcoming_ns = namespaces if (has_scope or not current_user.is_sec_team) else []
+        upcoming_escalations = await compute_upcoming_escalations(sx_db, app_db, upcoming_ns, settings)
+
     # Deduplicate by cve_id (same CVE can appear across multiple images).
     # Keep the entry with the highest epss_probability for each unique CVE.
     from datetime import datetime
@@ -191,6 +199,7 @@ async def team_dashboard(
     return TeamDashboardData(
         stat_total_cves=total,
         stat_escalations=escalations,
+        stat_upcoming_escalations=len(upcoming_escalations),
         stat_fixable_critical_cves=fixable_critical,
         stat_open_risk_acceptances=open_ra,
         severity_distribution=[