Creating Jira Tickets from Correlated SOC Alerts

pip install "aiohttp>=3.9,<4.0" "pydantic>=2.6,<3.0"

from __future__ import annotations

import logging
from typing import Awaitable, Callable, Optional, Protocol

from pydantic import BaseModel, Field

logger = logging.getLogger("soc.soar.jira")


class CorrelatedCase(BaseModel):
    case_id: str = Field(min_length=1)
    title: str = Field(min_length=1)
    severity: int = Field(ge=0, le=100)
    host_id: str = Field(min_length=1)
    technique_id: str | None = None


def to_issue_fields(case: CorrelatedCase, project_key: str) -> dict:
    priority = ("Highest" if case.severity >= 90 else
                "High" if case.severity >= 70 else
                "Medium" if case.severity >= 40 else "Low")
    return {
        "fields": {
            "project": {"key": project_key},
            "issuetype": {"name": "Security Incident"},
            "summary": case.title,
            "priority": {"name": priority},
            "labels": ["soc-auto", f"case-{case.case_id}",
                       case.technique_id or "no-attack"],
            "description": (f"Automated from correlated case {case.case_id}. "
                            f"Host {case.host_id}. ATT&CK {case.technique_id or 'n/a'}."),
        }
    }


class JiraApi(Protocol):
    async def search_by_label(self, label: str) -> Optional[str]: ...   # returns key
    async def create(self, payload: dict) -> str: ...
    async def update(self, key: str, payload: dict) -> None: ...


async def sync_issue(
    case: CorrelatedCase, api: JiraApi, project_key: str,
    claim: Callable[[str], Awaitable[bool]],
) -> str:
    payload = to_issue_fields(case, project_key)
    label = f"case-{case.case_id}"
    existing = await api.search_by_label(label)        # JQL: labels = "case-..."
    if existing is not None:
        await api.update(existing, payload)
        logger.info("jira updated case=%s key=%s", case.case_id, existing)
        return existing
    if not await claim(f"jira:{case.case_id}"):        # local race guard
        found = await api.search_by_label(label)       # reconcile
        if found:
            return found
    key = await api.create(payload)                    # ERR_TICKET_004 on failure
    logger.info("jira created case=%s key=%s", case.case_id, key)
    return key


if __name__ == "__main__":
    import asyncio

    class FakeJira:
        def __init__(self) -> None:
            self.created: list[dict] = []
            self.store: dict[str, str] = {}
        async def search_by_label(self, label: str):
            return self.store.get(label)
        async def create(self, payload: dict) -> str:
            key = f"SEC-{len(self.created) + 1}"
            self.created.append(payload)
            self.store[payload["fields"]["labels"][1]] = key
            return key
        async def update(self, key: str, payload: dict) -> None:
            pass

    async def claim(_k: str) -> bool:
        return True

    async def main() -> None:
        api = FakeJira()
        case = CorrelatedCase(case_id="C-9", title="Ransomware on FIN-DB-1",
                              severity=95, host_id="FIN-DB-1", technique_id="T1486")
        k1 = await sync_issue(case, api, "SEC", claim)
        k2 = await sync_issue(case, api, "SEC", claim)   # redelivery
        assert k1 == k2 and len(api.created) == 1

    asyncio.run(main())