"""Jira integration service. Authentication model -------------------- Each Aegis user authenticates to Jira with their own Atlassian email and personal API token. The email used is ``user.jira_email`` when set, falling back to ``user.email`` (the Aegis account email). This lets users specify a separate corporate Atlassian email without changing their Aegis login. The token is stored in ``user.jira_api_token``. Admin configuration ------------------- The Jira URL and default project key are stored in the ``system_configs`` table (keys ``jira.url`` and ``jira.project_key``) so the admin can update them at runtime without redeploying. These values override the legacy ``settings.JIRA_URL`` / ``settings.JIRA_DEFAULT_PROJECT`` env-vars which are kept for backwards-compatibility only. Lifecycle hooks --------------- ``push_test_event()`` is the single entry-point called from the test-workflow service on every state transition. It posts a rich comment to the linked Jira issue (if one exists) using the acting user's credentials. ``auto_create_test_issue()`` is called once after a test is created; it creates the Jira ticket and stores the link. """ from __future__ import annotations # Import logging import logging # Import re import re # Import datetime from datetime from datetime import datetime # Import Any, Optional from typing from typing import Optional # Import UUID from uuid from uuid import UUID # Import Session from sqlalchemy.orm from sqlalchemy.orm import Session # Import settings from app.config from app.config import settings # Import EntityNotFoundError from app.domain.errors from app.domain.errors import EntityNotFoundError # Import InvalidOperationError from app.domain.exceptions from app.domain.exceptions import InvalidOperationError # Import Campaign from app.models.campaign from app.models.campaign import Campaign # Import JiraLink, JiraLinkEntityType, JiraSyncDirection from app.models.jira_link from app.models.jira_link import JiraLink, JiraLinkEntityType, JiraSyncDirection # Import Technique from app.models.technique from app.models.technique import Technique # Import Test from app.models.test from app.models.test import Test from app.models.user import User # --------------------------------------------------------------------------- # Custom field IDs (project-specific — see System Settings → Jira Configuration) # --------------------------------------------------------------------------- JIRA_FIELD_RT_START_DATE = "customfield_11803" JIRA_FIELD_RT_END_DATE = "customfield_11804" JIRA_FIELD_BT_START_DATE = "customfield_11805" JIRA_FIELD_BT_END_DATE = "customfield_11806" JIRA_FIELD_TTP = "customfield_11807" JIRA_FIELD_SEVERITY = "customfield_10098" JIRA_FIELD_ATTACK_SUCCESS = "customfield_11815" JIRA_FIELD_ATTACK_DETECTED = "customfield_11809" JIRA_FIELD_TIME_TO_DETECT = "customfield_11810" JIRA_FIELD_TIME_TO_CONTAIN = "customfield_11811" JIRA_FIELD_DATA_SENSITIVITY = "customfield_11814" JIRA_FIELD_ATTACK_CONTAINED = "customfield_11885" _ATTACK_SUCCESS_TO_JIRA = { "successful": "Yes", "not_successful": "No", "partially_successful": "Partial", } _DETECTION_TO_JIRA = { "detected": "Yes", "not_detected": "No", "partially_detected": "Partial", } _CONTAINMENT_TO_JIRA = { "contained": "Yes", "not_contained": "No", "partially_contained": "Partial", } _DATA_CLASSIFICATION_TO_JIRA = { "public": "Public", "general_use": "General Use", "internal_use_only": "Internal Use Only", "trusted_people": "Trusted People", "customer_content": "Customer Content", "pii": "PII", } # Assign logger = logging.getLogger(__name__) logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # System-config helpers (admin-configurable Jira settings) # --------------------------------------------------------------------------- _JIRA_KEYS = { "url": "jira.url", "project_key": "jira.project_key", "enabled": "jira.enabled", } def _read_system_config(db: Session, key: str) -> Optional[str]: """Return a value from system_configs, or None if not set.""" 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 get_jira_url(db: Session) -> Optional[str]: """Return the admin-configured Jira URL, falling back to the env-var.""" return _read_system_config(db, _JIRA_KEYS["url"]) or settings.JIRA_URL or None def get_jira_project_key(db: Session) -> Optional[str]: """Return the admin-configured default project key, falling back to env-var.""" return ( _read_system_config(db, _JIRA_KEYS["project_key"]) or settings.JIRA_DEFAULT_PROJECT or None ) def is_jira_enabled(db: Session) -> bool: """Return True if Jira integration is enabled (DB setting or env-var).""" db_val = _read_system_config(db, _JIRA_KEYS["enabled"]) if db_val is not None: return db_val.lower() in ("true", "1", "yes") return settings.JIRA_ENABLED def get_jira_parent_ticket(db: Session) -> Optional[str]: """Return the configured parent ticket key for campaigns, or None if not set.""" return _read_system_config(db, "jira.parent_ticket") or None def get_jira_parent_ticket_standalone(db: Session) -> Optional[str]: """Return the parent ticket for standalone tests (not in a campaign). Falls back to get_jira_parent_ticket() if not explicitly configured. """ return ( _read_system_config(db, "jira.parent_ticket_standalone") or get_jira_parent_ticket(db) ) def upsert_jira_config(db: Session, key: str, value: str) -> None: """Persist a Jira config key-value pair.""" 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)) # --------------------------------------------------------------------------- # Per-user Jira client # --------------------------------------------------------------------------- def _effective_jira_email(user: User) -> Optional[str]: """Return the email to use for Jira auth: jira_email if set, otherwise email.""" return getattr(user, "jira_email", None) or user.email def get_user_jira_client(user: User, db: Session): """Build an Atlassian Jira client authenticated as *user*. Uses ``user.jira_email`` when set, otherwise falls back to ``user.email``. Raises ``InvalidOperationError`` when configuration is incomplete so callers can surface meaningful error messages. """ jira_url = get_jira_url(db) if not jira_url: raise InvalidOperationError( "Jira URL is not configured. Ask your administrator to set it in " "System Settings → Jira Configuration." ) auth_email = _effective_jira_email(user) if not auth_email: raise InvalidOperationError( "No email configured for Jira authentication. " "Set a Jira email in Settings → Profile → Jira Integration." ) if not user.jira_api_token: raise InvalidOperationError( "You have not configured a Jira API token. " "Go to Settings → Profile → Jira Integration and add your personal Atlassian token." ) from atlassian import Jira # Strip trailing slash — the Atlassian library appends paths like # /rest/api/2/myself and a trailing slash causes double-slash URLs. clean_url = jira_url.rstrip("/") return Jira( url=clean_url, username=auth_email, password=user.jira_api_token, cloud=True, ) def has_jira_configured(user: User, db: Session) -> bool: """Return True if *user* has everything needed to call Jira (legacy per-user check).""" return bool(get_jira_url(db) and _effective_jira_email(user) and user.jira_api_token) # --------------------------------------------------------------------------- # Admin Jira client (single account for all lifecycle hooks) # --------------------------------------------------------------------------- def get_admin_jira_email(db: Session) -> Optional[str]: """Return the admin Jira email from system_configs.""" return _read_system_config(db, "jira.admin_email") or None def get_admin_jira_api_token(db: Session) -> Optional[str]: """Return the admin Jira API token from system_configs.""" return _read_system_config(db, "jira.admin_api_token") or None def has_admin_jira_configured(db: Session) -> bool: """Return True if the global Jira admin account is fully configured.""" return bool( is_jira_enabled(db) and get_jira_url(db) and get_admin_jira_email(db) and get_admin_jira_api_token(db) ) def get_admin_jira_client(db: Session): """Return a Jira client authenticated with the global admin credentials. All Aegis-to-Jira operations (issue creation, comments, transitions) go through this single account. Users do not need personal Jira tokens. """ jira_url = get_jira_url(db) if not jira_url: raise InvalidOperationError("Jira URL is not configured.") admin_email = get_admin_jira_email(db) admin_token = get_admin_jira_api_token(db) if not admin_email or not admin_token: raise InvalidOperationError( "Admin Jira credentials not configured. " "Set them in System Settings → Jira Configuration → Admin Account." ) from atlassian import Jira return Jira( url=jira_url.rstrip("/"), username=admin_email, password=admin_token, cloud=True, ) def lookup_user_jira_account_id(db: Session, user: User) -> bool: """Lookup *user*'s Atlassian account ID by email using the admin Jira client. Uses ``user.jira_email`` when set (for users whose corporate Atlassian email differs from their Aegis login email), falling back to ``user.email`` — same resolution order as the rest of this module. Updates ``user.jira_account_id`` in-place when found or changed. Returns ``True`` when the value was updated, ``False`` otherwise. Non-fatal — all errors are logged at DEBUG level and swallowed. """ if not has_admin_jira_configured(db): return False email = getattr(user, "jira_email", None) or getattr(user, "email", None) if not email: return False try: jira = get_admin_jira_client(db) results = jira.user_find_by_user_string(query=email, limit=10) account_id: Optional[str] = None for u in results or []: if isinstance(u, dict) and u.get("emailAddress", "").lower() == email.lower(): account_id = u.get("accountId") break if not account_id: logger.debug("No Jira user found for email %s", email) return False current = getattr(user, "jira_account_id", None) if account_id != current: user.jira_account_id = account_id db.flush() logger.info( "Auto-updated jira_account_id for %s: %s", user.username, account_id ) return True return False except Exception as exc: logger.debug( "Could not lookup Jira account_id for %s: %s", getattr(user, "username", "?"), exc, ) return False def _resolve_jira_account_id(db: Session, user: Optional[User]) -> Optional[str]: """Return *user*'s Atlassian account ID, trying a fresh lookup if it isn't cached yet. ``jira_account_id`` is normally populated on login, but a user assigned or reviewing before their first login (or before Jira was configured) would otherwise leave every Jira assignment silently no-op'd forever — exactly the gap that caused a blue_lead's reviews to never reassign in Jira while red_lead's did, since only push_assignee_update had this fallback. Every Jira-assignment call site should use this, not read ``jira_account_id`` directly. """ if user is None: return None jira_account_id = getattr(user, "jira_account_id", None) if jira_account_id: return jira_account_id lookup_user_jira_account_id(db, user) db.flush() return getattr(user, "jira_account_id", None) # --------------------------------------------------------------------------- # Ticket content builders (inspired by the pentest-to-Jira script) # --------------------------------------------------------------------------- _SEVERITY_TO_PRIORITY: dict[str, str] = { "critical": "Highest", "high": "High", "medium": "Medium", "low": "Low", "informational": "Lowest", } _STATE_EMOJI: dict[str, str] = { "draft": "📝 Draft", "red_executing": "🔴 Red Team Executing", "red_review": "🔎 Red Lead Review", "blue_evaluating": "🔵 Blue Team Evaluating", "blue_review": "🔎 Blue Lead Review", "in_review": "📋 In Review", "validated": "✅ Validated", "rejected": "❌ Rejected", "disputed": "⚠️ Disputed", } # TestState -> Jira workflow status. These status names must already exist # as valid transition targets in the Purple Team project's Jira workflow # scheme (configured by a Jira admin) — set_issue_status() is a no-op # (logged warning, non-fatal) if the named status isn't a legal transition # from the issue's current status. # # "blue_evaluating" maps to "Queued Blue Team" here because that's the # state's *first-entry* meaning (approved by Red Lead, not yet claimed by # Blue, or re-routed to Blue Team during dispute resolution). When Blue # Lead sends work back to the *same* operator for rework # (reopen_blue_review), that's pushed as "In Progress" directly instead of # through this table — see push_bt_review_reopened(). _STATE_TO_JIRA_STATUS: dict[str, str] = { "draft": "To Do", "red_executing": "In Progress", "red_review": "Red Team test review", "blue_evaluating": "Queued Blue Team", "blue_review": "Blue Team test review", "in_review": "Validation", "validated": "Done", "rejected": "Rejected", "disputed": "Dispute", } def _technique_severity(technique: Optional[Technique]) -> str: """Return a lowercase severity string from the technique, defaulting to medium.""" if technique and hasattr(technique, "severity") and technique.severity: return technique.severity.lower() return "medium" def _build_test_description(test: Test, technique: Optional[Technique]) -> str: """Build the initial Jira ticket description for a newly created test.""" mitre_id = technique.mitre_id if technique else "N/A" tech_name = technique.name if technique else "N/A" tactic = technique.tactic if technique else "N/A" severity = _technique_severity(technique).capitalize() lines = [ "h2. Aegis Security Test", "", f"*Test Name:* {test.name}", f"*MITRE Technique:* [{mitre_id}|https://attack.mitre.org/techniques/{mitre_id.replace('.', '/')}] — {tech_name}", f"*Tactic:* {tactic}", f"*Platform:* {test.platform or 'N/A'}", f"*Severity:* {severity}", f"*Data Classification:* {test.data_classification or 'N/A'}", "", "h3. Description", test.description or "_No description provided._", "", f"*Tool:* {test.tool_used or 'N/A'}", "", "----", f"_Created via Aegis at {datetime.utcnow().strftime('%Y-%m-%d %H:%M')} UTC_", ] return "\n".join(lines) def _build_state_comment( test: Test, new_state: str, actor: User, extra: dict | None = None, ) -> str: """Build a Jira comment body for a test state transition.""" label = _STATE_EMOJI.get(new_state, new_state) lines = [ f"h3. {label}", "", f"*Changed by:* {actor.username} ({actor.email or 'no email'})", f"*At:* {datetime.utcnow().strftime('%Y-%m-%d %H:%M')} UTC", "", ] if new_state == "red_executing": lines += [ "Red Team has started the attack execution.", ] elif new_state == "red_review": lines += [ f"Red Team has submitted evidence for Round {test.red_round_number or 1} and the " "test is awaiting Red Lead review before it queues for Blue Team.", "", f"*Procedure:* {test.procedure_text or 'N/A'}", f"*Tool used:* {test.tool_used or 'N/A'}", f"*Attack Success:* {_enum_value(test.attack_success) or 'N/A'}", ] if test.red_summary: lines += ["", "h4. Red Team Summary", test.red_summary] elif new_state == "blue_review": lines += [ f"Blue Team has submitted evidence for Round {test.blue_round_number or 1} and the " "test is awaiting Blue Lead review before cross-validation.", "", f"*Detect Procedure:* {test.detect_procedure or 'N/A'}", f"*Detection Result:* {_enum_value(test.detection_result) or 'N/A'}", f"*Containment Result:* {_enum_value(test.containment_result) or 'N/A'}", ] if test.blue_summary: lines += ["", "h4. Blue Team Summary", test.blue_summary] elif new_state == "blue_evaluating": lines += [ "Red Team has finished execution and submitted evidence for Blue Team evaluation.", "", f"*Attack Success:* {_enum_value(test.attack_success) or 'N/A'}", ] if test.red_summary: lines += ["", "h4. Red Team Summary", test.red_summary] elif new_state == "in_review": lines += [ "Blue Team has completed evaluation. Test is awaiting lead validation.", "", f"*Detection Result:* {test.detection_result or 'N/A'}", ] if test.blue_summary: lines += ["", "h4. Blue Team Summary", test.blue_summary] if test.remediation_steps: lines += ["", "h4. Remediation Steps", test.remediation_steps] elif new_state == "validated": lines += [ "Test has been *validated* by both leads.", "", f"*Red Lead Status:* {test.red_validation_status or 'N/A'}", f"*Blue Lead Status:* {test.blue_validation_status or 'N/A'}", ] if test.red_validation_notes: lines += ["", f"*Red Lead Notes:* {test.red_validation_notes}"] if test.blue_validation_notes: lines += ["", f"*Blue Lead Notes:* {test.blue_validation_notes}"] elif new_state == "rejected": lines += [ "Test has been *rejected* and must be reworked.", "", f"*Red Lead Status:* {test.red_validation_status or 'N/A'}", f"*Blue Lead Status:* {test.blue_validation_status or 'N/A'}", ] if test.red_validation_notes: lines += ["", f"*Red Lead Notes:* {test.red_validation_notes}"] if test.blue_validation_notes: lines += ["", f"*Blue Lead Notes:* {test.blue_validation_notes}"] elif new_state == "draft": lines += ["Test has been reopened for re-execution."] # Any caller-supplied extra data if extra: lines.append("") for k, v in extra.items(): lines.append(f"*{k}:* {v}") lines.append("") lines.append("_Synced from [Aegis|https://aegis.undiamagico.es]_") return "\n".join(lines) # --------------------------------------------------------------------------- # Public lifecycle hooks # --------------------------------------------------------------------------- def _build_campaign_description(campaign) -> str: """Build the Jira ticket description for a campaign.""" lines = [ "h2. Aegis Security Campaign", "", f"*Campaign Name:* {campaign.name}", f"*Type:* {campaign.type}", f"*Status:* {campaign.status}", ] if campaign.description: lines += ["", "h3. Description", campaign.description] if campaign.tags: lines += ["", f"*Tags:* {', '.join(campaign.tags)}"] lines += [ "", "----", f"_Created via Aegis at {datetime.utcnow().strftime('%Y-%m-%d %H:%M')} UTC_", ] return "\n".join(lines) def get_campaign_jira_key(db: Session, campaign_id) -> Optional[str]: """Return the Jira issue key for a campaign, or None if not linked.""" import uuid as _uuid try: cid = _uuid.UUID(str(campaign_id)) except (ValueError, AttributeError): return None link = ( db.query(JiraLink) .filter( JiraLink.entity_type == JiraLinkEntityType.campaign, JiraLink.entity_id == cid, ) .first() ) return link.jira_issue_key if link else None def get_test_jira_key(db: Session, test_id) -> Optional[str]: """Return the Jira issue key for a test, or None if not linked.""" import uuid as _uuid try: tid = _uuid.UUID(str(test_id)) except (ValueError, AttributeError): return None link = ( db.query(JiraLink) .filter( JiraLink.entity_type == JiraLinkEntityType.test, JiraLink.entity_id == tid, ) .first() ) return link.jira_issue_key if link else None def auto_create_campaign_issue( db: Session, campaign, actor: User, ) -> Optional[str]: """Create a Jira issue for *campaign* under the configured parent ticket. Returns the Jira issue key on success, or ``None`` if Jira is not configured for *actor* or if the operation fails (non-fatal). Called once right after a campaign is committed to the database. The created ticket is stored as a JiraLink with entity_type=campaign. """ if not has_admin_jira_configured(db): return None project_key = get_jira_project_key(db) if not project_key: logger.warning( "Jira project key not configured; skipping auto-create for campaign %s", campaign.id, ) return None parent_ticket = get_jira_parent_ticket(db) try: jira = get_admin_jira_client(db) fields: dict = { "project": {"key": project_key}, "summary": f"[Aegis Campaign] {campaign.name}", "description": _build_campaign_description(campaign), "issuetype": {"name": settings.JIRA_ISSUE_TYPE_CAMPAIGN}, "labels": ["aegis", "campaign"], # customfield_10011 = Epic Name (required for Epic type in classic Jira) "customfield_10011": campaign.name, } # Set start date: use campaign.start_date if set, otherwise today effective_start = campaign.start_date or campaign.created_at if effective_start: fields[settings.JIRA_START_DATE_FIELD] = effective_start.strftime("%Y-%m-%d") # Nest under the configured parent ticket (Initiative, e.g. OFS-20795) if parent_ticket: fields["parent"] = {"key": parent_ticket} result = jira.issue_create(fields=fields) issue_key = result["key"] issue_id = result.get("id", "") link = JiraLink( entity_type=JiraLinkEntityType.campaign, entity_id=campaign.id, jira_issue_key=issue_key, jira_issue_id=issue_id, jira_project_key=project_key, sync_direction=JiraSyncDirection.aegis_to_jira, created_by=actor.id, ) db.add(link) db.flush() logger.info("Auto-created Jira issue %s for campaign %s", issue_key, campaign.id) return issue_key except Exception as exc: # Non-fatal: Jira failures must never break the campaign creation flow logger.warning( "Failed to auto-create Jira issue for campaign %s: %s", campaign.id, exc, exc_info=True, ) return None def auto_create_test_issue( db: Session, test: Test, actor: User, *, technique: Optional[Technique] = None, parent_ticket_override: Optional[str] = None, campaign_start_date=None, # datetime | None — inherited from campaign when available ) -> Optional[str]: """Create a Jira issue for *test* and store the link. Returns the Jira issue key on success, or ``None`` if Jira is not configured for *actor* or if the operation fails (non-fatal). Called once right after a test is committed to the database. Args: parent_ticket_override: When set, use this as the Jira parent ticket instead of the system-configured parent (e.g. OFS-9107). Use this to nest test tickets under a campaign ticket. """ if not has_admin_jira_configured(db): return None project_key = get_jira_project_key(db) if not project_key: logger.warning("Jira project key not configured; skipping auto-create for test %s", test.id) return None # Resolve technique if not supplied if technique is None: technique = db.query(Technique).filter(Technique.id == test.technique_id).first() mitre_id = technique.mitre_id if technique else "N/A" try: jira = get_admin_jira_client(db) # All tests — whether inside a campaign or standalone — are created # as Task. Campaign tests use the campaign Jira key as parent # (passed via parent_ticket_override); standalone tests use the # configured standalone parent ticket (e.g. OFS-20798, which is an # Epic so it can parent Tasks). parent = parent_ticket_override or get_jira_parent_ticket_standalone(db) issue_type = settings.JIRA_ISSUE_TYPE_TEST # always Task poc = test.procedure_text or "N/A" severity = _technique_severity(technique).capitalize() labels = ["aegis", mitre_id.replace(".", "-")] if test.platform: # Jira labels can't contain whitespace — normalize to kebab-case. labels.append(re.sub(r"[^a-z0-9_-]+", "-", test.platform.lower()).strip("-")) # No parent_ticket_override means this test isn't nested under a # campaign ticket — i.e. it's standalone. Tag it so standalone tests # are identifiable in Jira without cross-referencing Aegis. if parent_ticket_override is None: labels.append("standalone-test") fields: dict = { "project": {"key": project_key}, "summary": f"[Aegis] {mitre_id} — {test.name}", "description": _build_test_description(test, technique), "issuetype": {"name": issue_type}, "labels": labels, # customfield_10309 = Proof of Concept field (required by team's Jira config) "customfield_10309": f"{{code}}{poc}{{code}}", JIRA_FIELD_TTP: mitre_id, JIRA_FIELD_SEVERITY: _select_field(severity), } data_sensitivity = _DATA_CLASSIFICATION_TO_JIRA.get(_enum_value(test.data_classification)) if data_sensitivity: fields[JIRA_FIELD_DATA_SENSITIVITY] = _select_field(data_sensitivity) # Inherit campaign start date if available, otherwise use today from datetime import date as _date effective_start = campaign_start_date or _date.today() if hasattr(effective_start, "strftime"): fields[settings.JIRA_START_DATE_FIELD] = effective_start.strftime("%Y-%m-%d") if parent: fields["parent"] = {"key": parent} try: result = jira.issue_create(fields=fields) except Exception as exc: # Jira rejects the whole issue when a custom field isn't on the # project's create screen ("cannot be set ... not on the # appropriate screen"). Don't let a screen-config gap on an # optional field (data sensitivity) block ticket creation # entirely — drop it and retry once. if JIRA_FIELD_DATA_SENSITIVITY in fields and JIRA_FIELD_DATA_SENSITIVITY in str(exc): logger.warning( "Jira rejected %s (not on create screen); retrying without it for test %s", JIRA_FIELD_DATA_SENSITIVITY, test.id, ) fields.pop(JIRA_FIELD_DATA_SENSITIVITY) result = jira.issue_create(fields=fields) else: raise issue_key = result["key"] issue_id = result.get("id", "") link = JiraLink( entity_type=JiraLinkEntityType.test, entity_id=test.id, jira_issue_key=issue_key, jira_issue_id=issue_id, jira_project_key=project_key, sync_direction=JiraSyncDirection.aegis_to_jira, created_by=actor.id, ) db.add(link) db.flush() logger.info("Auto-created Jira issue %s for test %s", issue_key, test.id) return issue_key except Exception as exc: # Non-fatal: Jira failures must never break the test creation flow logger.warning( "Failed to auto-create Jira issue for test %s: %s", test.id, exc, exc_info=True, ) return None def push_test_event( db: Session, test: Test, actor: User, new_state: str, *, extra: dict | None = None, assignee: User | None = None, ) -> None: """Post a lifecycle comment to the Jira issue linked to *test*. Called from ``test_workflow_service`` after every state transition. Completely non-fatal — any Jira error is logged and swallowed so it never blocks the test workflow. """ if not has_admin_jira_configured(db): return link = ( db.query(JiraLink) .filter( JiraLink.entity_type == JiraLinkEntityType.test, JiraLink.entity_id == test.id, ) .first() ) if not link: return try: jira = get_admin_jira_client(db) comment = _build_state_comment(test, new_state, actor, extra) jira.issue_add_comment(link.jira_issue_key, comment) # Transition the Jira issue's status to match the new Aegis state, # per _STATE_TO_JIRA_STATUS. Covers the full Purple Team workflow: # To-Do -> In Progress -> RT Test Review -> Queued Blue Team -> # In Progress -> Blue Team Test Review -> Validation -> Done / # Rejected / Dispute. target_status = _STATE_TO_JIRA_STATUS.get(new_state) if target_status: try: jira.set_issue_status(link.jira_issue_key, target_status) logger.info( "Transitioned Jira ticket %s to %s", link.jira_issue_key, target_status ) except Exception as exc_t: logger.warning( "Could not transition %s to %s: %s", link.jira_issue_key, target_status, exc_t, ) # When the operator starts execution: assign the ticket to that # operator. On reopen (rework), *actor* is the lead who reopened # it, not the operator who should keep working it — the caller # passes the original red_tech_assignee as *assignee* in that case. if new_state == "red_executing": jira_account_id = _resolve_jira_account_id(db, assignee or actor) if jira_account_id: try: jira.assign_issue(link.jira_issue_key, account_id=jira_account_id) logger.info( "Assigned Jira ticket %s to account %s", link.jira_issue_key, jira_account_id, ) except Exception as exc_a: logger.warning( "Could not assign %s to %s: %s", link.jira_issue_key, jira_account_id, exc_a, ) if new_state in ("red_review", "blue_review") and assignee: jira_account_id = _resolve_jira_account_id(db, assignee) if jira_account_id: try: jira.assign_issue(link.jira_issue_key, account_id=jira_account_id) logger.info( "Assigned Jira ticket %s to reviewer account %s", link.jira_issue_key, jira_account_id, ) except Exception as exc_a: logger.warning( "Could not assign %s to reviewer %s: %s", link.jira_issue_key, jira_account_id, exc_a, ) # blue_evaluating is reached two different ways: a fresh hand-off # from Red (no blue_tech_assignee yet — queued, no owner) or a # reopen/rework of a test some blue_tech already had (assignee is # still set) — that operator should get the ticket back, not have # it clear. in_review is always dual-validation by both leads at # once, so it never has a single owner. if new_state == "blue_evaluating": if test.blue_tech_assignee: reassignee = db.query(User).filter(User.id == test.blue_tech_assignee).first() jira_account_id = _resolve_jira_account_id(db, reassignee) else: jira_account_id = None try: jira.assign_issue(link.jira_issue_key, account_id=jira_account_id) logger.info( "%s Jira ticket %s (state=%s)", "Reassigned" if jira_account_id else "Unassigned", link.jira_issue_key, new_state, ) except Exception as exc_a: logger.warning( "Could not update assignee on %s for state %s: %s", link.jira_issue_key, new_state, exc_a, ) elif new_state == "in_review": try: jira.assign_issue(link.jira_issue_key, account_id=None) logger.info("Unassigned Jira ticket %s (state=%s)", link.jira_issue_key, new_state) except Exception as exc_a: logger.warning( "Could not unassign %s for state %s: %s", link.jira_issue_key, new_state, exc_a, ) link.last_synced_at = datetime.utcnow() db.flush() logger.info( "Posted Jira comment to %s for test %s state=%s", link.jira_issue_key, test.id, new_state, ) except Exception as exc: logger.warning( "Failed to push Jira event for test %s (state=%s): %s", test.id, new_state, exc, exc_info=True, ) def push_assignee_update(db: Session, test: Test, assignee: User) -> None: """Sync a lead's manual operator assignment to the linked Jira ticket. Called from ``POST /tests/{id}/assign`` so Jira reflects the assignment immediately, instead of waiting for the operator to start execution (the only other point that pushes a Jira assignee — see ``push_test_event``'s ``red_executing`` handling above). """ if not has_admin_jira_configured(db): return link = ( db.query(JiraLink) .filter( JiraLink.entity_type == JiraLinkEntityType.test, JiraLink.entity_id == test.id, ) .first() ) if not link: return jira_account_id = _resolve_jira_account_id(db, assignee) if not jira_account_id: return try: jira = get_admin_jira_client(db) jira.assign_issue(link.jira_issue_key, account_id=jira_account_id) link.last_synced_at = datetime.utcnow() db.flush() logger.info( "Assigned Jira ticket %s to account %s (manual assignment)", link.jira_issue_key, jira_account_id, ) except Exception as exc: logger.warning( "Could not assign %s to %s: %s", link.jira_issue_key, jira_account_id, exc, ) # --------------------------------------------------------------------------- # On-hold Jira notification # --------------------------------------------------------------------------- def push_hold_event( db: Session, test, actor, *, resuming: bool = False, reason: str | None = None, ) -> None: """Post an on-hold / resume comment to the Jira issue linked to *test*. Non-fatal — any Jira error is logged and swallowed. """ if not has_admin_jira_configured(db): return link = ( db.query(JiraLink) .filter( JiraLink.entity_type == JiraLinkEntityType.test, JiraLink.entity_id == test.id, ) .first() ) if not link: return try: jira = get_admin_jira_client(db) ts = datetime.utcnow().strftime("%Y-%m-%d %H:%M UTC") if resuming: comment = ( f"h3. ▶ Test Resumed\n\n" f"*Resumed by:* {actor.username}\n" f"*At:* {ts}\n\n" f"_The test has been unblocked and is active again._" ) try: jira.set_issue_status(link.jira_issue_key, "In Progress") except Exception as exc_t: logger.warning("Could not transition %s off blocked: %s", link.jira_issue_key, exc_t) else: comment = ( f"h3. 🚫 Test Blocked\n\n" f"*Blocked by:* {actor.username}\n" f"*At:* {ts}\n" f"*Reason:* {reason or 'No reason provided'}\n\n" f"_This test is blocked. No action required until it is resumed._" ) try: jira.set_issue_status(link.jira_issue_key, "Blocked") except Exception as exc_t: logger.warning("Could not transition %s to Blocked: %s", link.jira_issue_key, exc_t) jira.issue_add_comment(link.jira_issue_key, comment) link.last_synced_at = datetime.utcnow() db.flush() logger.info("Posted hold event to Jira %s (resuming=%s)", link.jira_issue_key, resuming) except Exception as exc: logger.warning("Failed to push hold event for test %s: %s", test.id, exc, exc_info=True) def push_pause_event(db: Session, test, actor, *, resuming: bool = False) -> None: """Post a timer pause/resume comment to Jira — status is left untouched. Distinct from :func:`push_hold_event` — a short timer pause (the timer sitting idle between rework rounds, or the operator briefly stepping away) is not the same thing as the test actually being blocked, so it must not flip the Jira issue to "On Hold". Only a genuine Hold action (:func:`push_hold_event`) does that. This just leaves a record. Non-fatal — any Jira error is logged and swallowed. """ if not has_admin_jira_configured(db): return link = ( db.query(JiraLink) .filter( JiraLink.entity_type == JiraLinkEntityType.test, JiraLink.entity_id == test.id, ) .first() ) if not link: return try: jira = get_admin_jira_client(db) ts = datetime.utcnow().strftime("%Y-%m-%d %H:%M UTC") if resuming: comment = f"h3. ▶ Timer Resumed\n\n*Resumed by:* {actor.username}\n*At:* {ts}" else: comment = f"h3. ⏸ Timer Paused\n\n*Paused by:* {actor.username}\n*At:* {ts}" jira.issue_add_comment(link.jira_issue_key, comment) link.last_synced_at = datetime.utcnow() db.flush() logger.info("Posted pause event to Jira %s (resuming=%s)", link.jira_issue_key, resuming) except Exception as exc: logger.warning("Failed to push pause event for test %s: %s", test.id, exc, exc_info=True) def push_round_archived(db: Session, test, actor, *, round_data) -> None: """Post a permanent Jira comment summarizing a round right before it's reset for rework. ``round_data`` custom fields (attack success, detection result, etc.) only ever show Jira's *latest* value — each new round's submission overwrites the field. This comment is what preserves every prior round's results in Jira, since comments are never overwritten. Non-fatal — any Jira error is logged and swallowed. """ if not has_admin_jira_configured(db): return link = ( db.query(JiraLink) .filter( JiraLink.entity_type == JiraLinkEntityType.test, JiraLink.entity_id == test.id, ) .first() ) if not link: return try: jira = get_admin_jira_client(db) team = round_data.team lines = [f"h3. \U0001F4CB Round {round_data.round_number} archived ({team} team)\n"] if team == "red": lines.append(f"*Procedure:* {round_data.procedure_text or '_none recorded_'}") lines.append(f"*Tool used:* {round_data.tool_used or '-'}") lines.append(f"*Attack success:* {round_data.attack_success.value if round_data.attack_success else '-'}") lines.append(f"*Summary:* {round_data.red_summary or '-'}") else: lines.append(f"*Detect procedure:* {round_data.detect_procedure or '_none recorded_'}") lines.append(f"*Detection result:* {round_data.detection_result.value if round_data.detection_result else '-'}") lines.append(f"*Containment result:* {round_data.containment_result.value if round_data.containment_result else '-'}") lines.append(f"*Summary:* {round_data.blue_summary or '-'}") lines.append(f"*Sent back by:* {actor.username}") lines.append(f"*Reopen notes:* {round_data.review_notes or '-'}") comment = "\n".join(lines) jira.issue_add_comment(link.jira_issue_key, comment) _add_jira_label(jira, link.jira_issue_key, "reopened") link.last_synced_at = datetime.utcnow() db.flush() logger.info( "Posted round-archived comment to Jira %s (team=%s, round=%s)", link.jira_issue_key, team, round_data.round_number, ) except Exception as exc: logger.warning("Failed to push round-archived comment for test %s: %s", test.id, exc, exc_info=True) def _add_jira_label(jira, issue_key: str, label: str) -> None: """Add *label* to an issue's labels without clobbering the existing ones. Jira's update API takes the whole ``labels`` array, so adding one means read-modify-write: fetch the current list, append if missing, write it back. Failures here are logged and swallowed by the caller. """ issue = jira.issue(issue_key, fields="labels") current = (issue.get("fields") or {}).get("labels") or [] if label not in current: jira.update_issue_field(issue_key, fields={"labels": current + [label]}) def push_bt_work_started(db: Session, test, actor) -> None: """Transition the Jira issue to "In Progress" when a blue tech picks up a queued test to start evaluating. The test's TestState stays "blue_evaluating" throughout (queued AND actively worked are the same Aegis state), so this can't be driven by the generic push_test_event()/_STATE_TO_JIRA_STATUS dispatch — that would incorrectly re-push "Queued Blue Team". Non-fatal. """ if not has_admin_jira_configured(db): return link = ( db.query(JiraLink) .filter( JiraLink.entity_type == JiraLinkEntityType.test, JiraLink.entity_id == test.id, ) .first() ) if not link: return try: jira = get_admin_jira_client(db) ts = datetime.utcnow().strftime("%Y-%m-%d %H:%M UTC") comment = f"h3. ▶ Blue Team Started\n\n*Picked up by:* {actor.username}\n*At:* {ts}" try: jira.set_issue_status(link.jira_issue_key, "In Progress") except Exception as exc_t: logger.warning("Could not transition %s to In Progress: %s", link.jira_issue_key, exc_t) jira.issue_add_comment(link.jira_issue_key, comment) jira_account_id = getattr(actor, "jira_account_id", None) if jira_account_id: try: jira.assign_issue(link.jira_issue_key, account_id=jira_account_id) except Exception as exc_a: logger.warning( "Could not assign %s to %s: %s", link.jira_issue_key, jira_account_id, exc_a, ) link.last_synced_at = datetime.utcnow() db.flush() logger.info("Posted blue-team-started event to Jira %s", link.jira_issue_key) except Exception as exc: logger.warning("Failed to push blue-team-started event for test %s: %s", test.id, exc, exc_info=True) # --------------------------------------------------------------------------- # RT/BT date + result custom-field sync # --------------------------------------------------------------------------- def _enum_value(v) -> Optional[str]: """Return the plain string value of an enum member, or the value itself.""" if v is None: return None return v.value if hasattr(v, "value") else str(v) def _select_field(value: str) -> dict: """Wrap a value for a Jira single-select custom field. Jira's REST API rejects a bare string for select-type custom fields ("Specify a valid 'id' or 'name' for ") — it must be posted as ``{"value": ...}``. """ return {"value": value} def _jira_datetime(dt: datetime) -> str: """Format a naive-UTC datetime for a Jira ``datetime``-type custom field. RT/BT Start/End Date are genuine Jira datetime fields (confirmed via issue_editmeta — schema type "datetime"), not plain dates. Sending a bare "YYYY-MM-DD" string makes Jira default the time-of-day to midnight, so RT Start Date and RT End Date ended up showing the same meaningless midnight timestamp regardless of when the operator actually started/finished. Jira's REST API expects ``yyyy-MM-dd'T'HH:mm:ss.SSSZ`` for this field type. """ return dt.strftime("%Y-%m-%dT%H:%M:%S.000+0000") def _update_test_fields(db: Session, test: Test, fields: dict) -> None: """Best-effort update of custom fields on the Jira issue linked to *test*. Non-fatal — any Jira error is logged and swallowed so it never blocks the test workflow. """ if not fields or not has_admin_jira_configured(db): return link = ( db.query(JiraLink) .filter( JiraLink.entity_type == JiraLinkEntityType.test, JiraLink.entity_id == test.id, ) .first() ) if not link: return try: jira = get_admin_jira_client(db) jira.update_issue_field(link.jira_issue_key, fields=fields) link.last_synced_at = datetime.utcnow() db.flush() logger.info("Updated Jira fields %s on %s", list(fields.keys()), link.jira_issue_key) except Exception as exc: logger.warning( "Failed to update Jira fields for test %s: %s", test.id, exc, exc_info=True ) def push_rt_submitted(db: Session, test: Test) -> None: """Set RT Start Date, RT End Date, and Attack Success — called when Red submits for review. RT Start Date always reflects the *first* round's execution start (from round_number=1 in round_history if this test has been reopened, else the live field — this is round 1 in that case). RT End Date and Attack Success always reflect the *current* round, since they're read live off the test each time this runs and get pushed again on every resubmit. Previously this pushed ``datetime.utcnow()`` (click time) instead of the operator-entered execution window — fixed to read the real fields. """ from app.models.test_round_history import TestRoundHistory fields: dict = {} first_round = ( db.query(TestRoundHistory) .filter( TestRoundHistory.test_id == test.id, TestRoundHistory.team == "red", TestRoundHistory.round_number == 1, ) .first() ) first_start = ( first_round.execution_start_time if first_round and first_round.execution_start_time else test.execution_start_time ) if first_start: fields[JIRA_FIELD_RT_START_DATE] = _jira_datetime(first_start) if test.execution_end_time: fields[JIRA_FIELD_RT_END_DATE] = _jira_datetime(test.execution_end_time) attack_success = _enum_value(test.attack_success) if attack_success in _ATTACK_SUCCESS_TO_JIRA: fields[JIRA_FIELD_ATTACK_SUCCESS] = _select_field(_ATTACK_SUCCESS_TO_JIRA[attack_success]) _update_test_fields(db, test, fields) def push_bt_started(db: Session, test: Test) -> None: """Set the BT Start Date field — called when Blue picks up the test to evaluate. Only pushed on round 1. On later rounds (after a reopen), the field already holds round 1's real pickup date and must not be overwritten — same "first round survives every reopen" rule as RT Start Date. """ if (test.blue_round_number or 1) > 1: return _update_test_fields(db, test, { JIRA_FIELD_BT_START_DATE: _jira_datetime(datetime.utcnow()), }) def push_bt_submitted(db: Session, test: Test) -> None: """Set BT End Date, Attack Detected/Contained, and Time to Detect/Contain — Blue submits for review. Time to Detect / Time to Contain are Jira **datetime** fields (confirmed via issue_editmeta), not numeric duration fields — despite the name. This used to push a computed hour count, which Jira rejected outright; since Jira validates the whole fields payload atomically, that one bad field silently sank BT End Date and Attack Detected too. Now pushes the actual detection_time/containment_time timestamps. """ fields: dict = {JIRA_FIELD_BT_END_DATE: _jira_datetime(datetime.utcnow())} detection_result = _enum_value(test.detection_result) if detection_result in _DETECTION_TO_JIRA: fields[JIRA_FIELD_ATTACK_DETECTED] = _select_field(_DETECTION_TO_JIRA[detection_result]) containment_result = _enum_value(test.containment_result) if containment_result in _CONTAINMENT_TO_JIRA: fields[JIRA_FIELD_ATTACK_CONTAINED] = _select_field(_CONTAINMENT_TO_JIRA[containment_result]) if test.detection_time: fields[JIRA_FIELD_TIME_TO_DETECT] = _jira_datetime(test.detection_time) if test.containment_time: fields[JIRA_FIELD_TIME_TO_CONTAIN] = _jira_datetime(test.containment_time) _update_test_fields(db, test, fields) # --------------------------------------------------------------------------- # Legacy / generic helpers (kept for existing routes) # --------------------------------------------------------------------------- def get_jira_client(): """Return a shared Jira client using global credentials (legacy path). Raises ``InvalidOperationError`` when Jira is disabled or unconfigured. Prefer ``get_user_jira_client()`` for new code. """ if not settings.JIRA_ENABLED: # Raise InvalidOperationError raise InvalidOperationError("Jira integration is not enabled") if not settings.JIRA_URL or not settings.JIRA_USERNAME or not settings.JIRA_API_TOKEN: raise InvalidOperationError( "Jira is enabled but JIRA_URL / JIRA_USERNAME / JIRA_API_TOKEN are not set" ) from atlassian import Jira return Jira( url=settings.JIRA_URL, username=settings.JIRA_USERNAME, password=settings.JIRA_API_TOKEN, cloud=settings.JIRA_IS_CLOUD, ) # Define function search_jira_issues def search_jira_issues(query: str, max_results: int = 10) -> list[dict]: """Search Jira issues by JQL or free text (uses global credentials).""" jira = get_jira_client() # Assign jql = query if "=" in query or "~" in query else f'summary ~ "{query}"' jql = query if "=" in query or "~" in query else f'summary ~ "{query}"' # Assign results = jira.jql(jql, limit=max_results) results = jira.jql(jql, limit=max_results) # Return [ return [ { # Literal argument value "issue_key": issue["key"], # Literal argument value "summary": issue["fields"]["summary"], # Literal argument value "status": issue["fields"]["status"]["name"], # Literal argument value "assignee": (issue["fields"].get("assignee") or {}).get("displayName"), # Literal argument value "priority": (issue["fields"].get("priority") or {}).get("name"), } for issue in results.get("issues", []) ] # Define function create_jira_issue def create_jira_issue( # Entry: project_key project_key: str, # Entry: summary summary: str, # Entry: description description: str, # Entry: issue_type issue_type: str = "Task", # Entry: labels labels: Optional[list[str]] = None, # Entry: custom_fields custom_fields: Optional[dict] = None, ) -> dict: """Create a Jira issue and return its key + id (uses global credentials).""" jira = get_jira_client() # Assign fields = { fields: dict = { # Literal argument value "project": {"key": project_key}, # Literal argument value "summary": summary, # Literal argument value "description": description, # Literal argument value "issuetype": {"name": issue_type}, } # Check: labels if labels: # Assign fields["labels"] = labels fields["labels"] = labels # Check: custom_fields if custom_fields: # Call fields.update() fields.update(custom_fields) # Assign result = jira.issue_create(fields=fields) result = jira.issue_create(fields=fields) # Return {"issue_key": result["key"], "issue_id": result["id"]} return {"issue_key": result["key"], "issue_id": result["id"]} # Define function sync_jira_to_aegis def sync_jira_to_aegis(db: Session, link: JiraLink) -> None: """Pull current status from Jira into the local link record (global creds).""" jira = get_jira_client() # Assign issue = jira.issue(link.jira_issue_key) issue = jira.issue(link.jira_issue_key) # Assign fields = issue.get("fields", {}) fields = issue.get("fields", {}) # Assign link.jira_status = fields.get("status", {}).get("name") link.jira_status = fields.get("status", {}).get("name") # Assign link.jira_priority = (fields.get("priority") or {}).get("name") link.jira_priority = (fields.get("priority") or {}).get("name") # Assign link.jira_assignee = (fields.get("assignee") or {}).get("displayName") link.jira_assignee = (fields.get("assignee") or {}).get("displayName") # Assign link.jira_story_points = str(fields.get("customfield_10016", "")) link.jira_story_points = str(fields.get("customfield_10016", "")) # Assign link.last_synced_at = datetime.utcnow() link.last_synced_at = datetime.utcnow() # Flush changes to DB without committing the transaction db.flush() # Define function sync_aegis_to_jira def sync_aegis_to_jira(db: Session, link: JiraLink, entity_data: dict) -> None: """Push an Aegis status update as a Jira comment (global creds).""" jira = get_jira_client() # Assign comment_body = _build_sync_comment(entity_data) comment_body = _build_sync_comment(entity_data) # Call jira.issue_add_comment() jira.issue_add_comment(link.jira_issue_key, comment_body) # Assign link.last_synced_at = datetime.utcnow() link.last_synced_at = datetime.utcnow() # Flush changes to DB without committing the transaction db.flush() # Define function _build_sync_comment def _build_sync_comment(data: dict) -> str: lines = ["h3. Aegis Sync Update", ""] # Iterate over data.items() for key, value in data.items(): # Call lines.append() lines.append(f"*{key}:* {value}") # Call lines.append() lines.append(f"\n_Synced at {datetime.utcnow().isoformat()}_") # Return "\n".join(lines) return "\n".join(lines) # ── Link CRUD ──────────────────────────────────────────────────────────── def create_link( # Entry: db db: Session, *, # Entry: entity_type entity_type: JiraLinkEntityType, # Entry: entity_id entity_id: UUID, # Entry: jira_issue_key jira_issue_key: str, # Entry: sync_direction sync_direction: JiraSyncDirection, # Entry: created_by created_by: UUID, ) -> JiraLink: link = JiraLink( # Keyword argument: entity_type entity_type=entity_type, # Keyword argument: entity_id entity_id=entity_id, # Keyword argument: jira_issue_key jira_issue_key=jira_issue_key, # Keyword argument: sync_direction sync_direction=sync_direction, # Keyword argument: created_by created_by=created_by, ) # Stage new record(s) for database insertion db.add(link) # Flush changes to DB without committing the transaction db.flush() # Check: settings.JIRA_ENABLED if settings.JIRA_ENABLED: # Attempt the following; catch errors below try: # Call sync_jira_to_aegis() sync_jira_to_aegis(db, link) # Handle Exception except Exception as e: # Log warning: "Initial Jira sync failed for %s: %s", jira_issue_ logger.warning("Initial Jira sync failed for %s: %s", jira_issue_key, e) # Return link return link # Define function list_links def list_links( # Entry: db db: Session, *, # Entry: entity_type entity_type: Optional[JiraLinkEntityType] = None, # Entry: entity_id entity_id: Optional[UUID] = None, entity_ids: Optional[list[UUID]] = None, ) -> list[JiraLink]: query = db.query(JiraLink) # Check: entity_type if entity_type: # Assign query = query.filter(JiraLink.entity_type == entity_type) query = query.filter(JiraLink.entity_type == entity_type) # Check: entity_id if entity_id: # Assign query = query.filter(JiraLink.entity_id == entity_id) query = query.filter(JiraLink.entity_id == entity_id) elif entity_ids: query = query.filter(JiraLink.entity_id.in_(entity_ids)) return query.order_by(JiraLink.created_at.desc()).all() # Define function get_link_or_raise def get_link_or_raise(db: Session, link_id: UUID) -> JiraLink: link = db.query(JiraLink).filter(JiraLink.id == link_id).first() # Check: not link if not link: # Raise EntityNotFoundError raise EntityNotFoundError("JiraLink", str(link_id)) # Return link return link # Define function delete_link def delete_link(db: Session, link_id: UUID) -> JiraLink: link = get_link_or_raise(db, link_id) # Mark record for deletion on next commit db.delete(link) # Return link return link def build_issue_data( db: Session, entity_type: JiraLinkEntityType, entity_id: UUID ) -> tuple[str, str]: """Build Jira issue summary and description from an Aegis entity.""" # Check: entity_type == JiraLinkEntityType.test if entity_type == JiraLinkEntityType.test: # Assign entity = db.query(Test).filter(Test.id == entity_id).first() entity = db.query(Test).filter(Test.id == entity_id).first() # Check: not entity if not entity: # Raise EntityNotFoundError raise EntityNotFoundError("Test", str(entity_id)) technique = db.query(Technique).filter(Technique.id == entity.technique_id).first() return ( f"[Aegis] {technique.mitre_id if technique else 'N/A'} — {entity.name}", _build_test_description(entity, technique), ) # Alternative: entity_type == JiraLinkEntityType.campaign elif entity_type == JiraLinkEntityType.campaign: # Assign entity = db.query(Campaign).filter(Campaign.id == entity_id).first() entity = db.query(Campaign).filter(Campaign.id == entity_id).first() # Check: not entity if not entity: # Raise EntityNotFoundError raise EntityNotFoundError("Campaign", str(entity_id)) # Return ( return ( f"[Aegis Campaign] {entity.name}", f"Campaign: {entity.name}\nType: {entity.type}\nStatus: {entity.status}\n" f"Description: {entity.description or 'N/A'}", ) # Alternative: entity_type == JiraLinkEntityType.technique elif entity_type == JiraLinkEntityType.technique: # Assign entity = db.query(Technique).filter(Technique.id == entity_id).first() entity = db.query(Technique).filter(Technique.id == entity_id).first() # Check: not entity if not entity: # Raise EntityNotFoundError raise EntityNotFoundError("Technique", str(entity_id)) # Return ( return ( f"[Aegis Technique] {entity.mitre_id} - {entity.name}", f"MITRE ID: {entity.mitre_id}\nName: {entity.name}\n" f"Tactic: {entity.tactic or 'N/A'}\n" f"Description: {entity.description or 'N/A'}", ) # Fallback: handle remaining cases else: # Return f"[Aegis] Entity {entity_id}", f"Entity type: {entity_type.value}" return f"[Aegis] Entity {entity_id}", f"Entity type: {entity_type.value}" # Define function create_issue_and_link def create_issue_and_link( # Entry: db db: Session, *, # Entry: entity_type entity_type: JiraLinkEntityType, # Entry: entity_id entity_id: UUID, # Entry: created_by created_by: UUID, ) -> dict: """Create a Jira issue from an Aegis entity and link them (global creds).""" summary, description = build_issue_data(db, entity_type, entity_id) project_key = settings.JIRA_DEFAULT_PROJECT result = create_jira_issue( project_key=project_key, summary=summary, # Keyword argument: description description=description, # Keyword argument: labels labels=["aegis", entity_type.value], ) # Assign link = JiraLink( link = JiraLink( # Keyword argument: entity_type entity_type=entity_type, # Keyword argument: entity_id entity_id=entity_id, # Keyword argument: jira_issue_key jira_issue_key=result["issue_key"], # Keyword argument: jira_issue_id jira_issue_id=result["issue_id"], jira_project_key=project_key, created_by=created_by, ) # Stage new record(s) for database insertion db.add(link) # Return {"issue_key": result["issue_key"], "link_id": str(link.id)} return {"issue_key": result["issue_key"], "link_id": str(link.id)}