ReliaQuest GreyMatter DRP Incidents

from __future__ import annotations

import hashlib
import random

import pytest
from CommonServerPython import *
from ReliaQuestGreyMatterDRPIncidents import Client

from Packs.DigitalShadows.Integrations.ReliaQuestGreyMatterDRPIncidents.ReliaQuestGreyMatterDRPIncidents import search_find

TEST_URL = "https://test.com/api"

RISK_TYPES = [
    "exposed-credential",
    "impersonating-domain",
    "impersonating-subdomain",
    "unauthorized-code-commit",
    "exposed-access-key",
]


@pytest.fixture
def client() -> Client:
    return Client(
        base_url=TEST_URL, account_id="1234", access_key="HHHQW", secret_key="saddsadsajhhjksa", verify=False, proxy=False
    )


def create_alert_with_risk_type(id, risk_type):
    return [
        {
            "id": _id,
            "portal-id": "BFM9J",
            "risk-type": risk_type,
            "classification": "exposed-credential-alert",
            "risk-assessment": {"risk-level": "low"},
            "risk-factors": ["Exposed on open source", "Plain text password", "More than a year old when raised"],
            "title": "The alert title",
            "description": "A description of this alert",
            "assets": [{"id": "1de8c226-16ef-4c44-a2b4-11a769e6b377"}],
            "raised": "2020-04-01T08:30:00Z",
            "updated": "2020-04-01T08:30:00Z",
            "email": "username@example.org",
            "password": "P@ssw0rd",
            "inferred-password-type": "scrypt",
            "first-seen": "2020-04-01T08:30:00Z",
        }
        for _id in id
    ]


def create_search_find_response():
    return {"content": [], "total": 0}


class ClientMock:
    LAST_EVENT_TIME_AND_NUM = None

    def __init__(self, num_of_events: int, num_of_alerts: int, num_of_incidents: int):
        self.num_of_events = num_of_events
        self.num_of_fetched_events = 0
        self.num_of_alerts = num_of_alerts
        self.num_of_fetched_alerts = 0
        self.num_of_incidents = num_of_incidents
        self.num_of_fetched_incidents = 0

    def http_request_side_effect(self, method: str, url_suffix: str, params: Dict | None = None, **kwargs):
        if url_suffix == "/v1/triage-item-events":
            if self.num_of_fetched_events >= self.num_of_events:
                return create_mocked_response([])
            limit = params["limit"]
            event_num_after = params["event-num-after"]
            if event_num_after:
                event_num_after += 1
            response = create_triage_item_events(limit, start_event_num=event_num_after or 1)
            self.num_of_fetched_events += len(response)
            if self.num_of_fetched_events > self.num_of_events:
                response = response[: self.num_of_fetched_events - self.num_of_events]

        elif url_suffix == "/v1/triage-items":
            triage_item_ids = params["id"]
            triage_item_alerts = triage_item_ids[: self.num_of_alerts - self.num_of_fetched_alerts]
            alerts_response = create_triage_items_from_events(triage_item_alerts, item_type="alert-id")
            triage_item_incidents = triage_item_ids[self.num_of_alerts - self.num_of_fetched_alerts :]
            incidents_response = create_triage_items_from_events(triage_item_incidents, item_type="incident-id")

            self.num_of_fetched_alerts += len(alerts_response)
            self.num_of_fetched_incidents += len(incidents_response)

            response = alerts_response + incidents_response

        elif url_suffix == "/v1/alerts":
            response = create_incidents_and_alerts_from_triaged_items(params["id"], item_type="alert-id", amount_of_assets=1)
        elif url_suffix == "/v1/incidents":
            response = create_incidents_and_alerts_from_triaged_items(params["id"], item_type="incident-id", amount_of_assets=1)
        elif url_suffix == "/v1/assets":
            response = create_assets(params["id"])
        elif url_suffix == "/v1/triage-item-comments":
            response = create_comments(params["id"])
        elif url_suffix in [
            "/v1/exposed-credential-alerts",
            "/v1/impersonating-domain-alerts",
            "/v1/impersonating-subdomain-alerts",
            "/v1/unauthorized-code-commit-alerts",
            "/v1/exposed-access-key-alerts",
        ]:
            response = create_alert_with_risk_type(params["id"], url_suffix.split("/")[-1])
        elif url_suffix == "/api/search/find":
            response = create_search_find_response()
        else:
            response = []

        return create_mocked_response(response)


def create_mocked_response(response: List[Dict] | Dict, status_code: int = 200) -> requests.Response:
    mocked_response = requests.Response()
    mocked_response._content = json.dumps(response).encode("utf-8")
    mocked_response.status_code = status_code
    return mocked_response


def create_triage_item_events(num_of_events: int, start_event_num: int = 1) -> List[Dict]:
    return [
        {
            "event-num": event_num,
            "event-created": "2020-09-24T16:30:10.016Z",
            "triage-item-id": event_num,
            "event-action": "create",
            "risk-level": "high",
            "risk-type": "test",
            "classification": "test",
            "state": "unread",
        }
        for event_num in range(start_event_num, num_of_events + start_event_num)
    ]


def create_triage_items_from_events(triage_item_ids: List[str], item_type: str) -> List[Dict]:
    if item_type not in {"incident-id", "alert-id"}:
        raise ValueError(f"item-type {item_type} must be one of incident-id/alert-id")

    return [
        {
            "id": triage_item_id,
            "title": "title",
            "state": "unread",
            "raised": "date",
            "source": {item_type: triage_item_id},
            "risk-type": random.choice(RISK_TYPES),
        }
        for triage_item_id in triage_item_ids
    ]


def create_incidents_and_alerts_from_triaged_items(_ids: List[str], item_type: str, amount_of_assets: int = 0) -> List[Dict]:
    if item_type not in {"incident-id", "alert-id"}:
        raise ValueError(f"item-type {item_type} must be one of incident-id/alert-id")
    events = []
    for _id in _ids:
        events.append({"id": _id, "title": f"{item_type}-{_id}", "assets": []})

    if amount_of_assets > 0:
        for i, event in enumerate(events):
            event_copy = event.copy()
            event_copy["unique_id"] = i
            event["assets"].append({"id": hashlib.sha256(json.dumps(event_copy, sort_keys=True).encode()).hexdigest()})

    return events


def create_assets(asset_ids: List[str]) -> List[Dict]:
    return [{"id": _id, "type": f"asset-{_id}", "approval-state": "accepted", "display-value": "test"} for _id in asset_ids]


def create_comments(comment_ids):
    return [
        {
            "id": _id,
            "triage-item-id": _id,
            "content": f"test content-{_id}",
            "updated": "2020-04-01T08:30:00Z",
            "created": "2020-04-01T08:30:00Z",
            "user": None,
        }
        for _id in comment_ids
    ]


def test_fetch_incidents_command(mocker, client: Client):
    """
    Given:
     - 100 events

    When:
     - running the fetch_incidents_command

    Then:
     - make sure that all events are enriched and fetched (5000)
    """
    from ReliaQuestGreyMatterDRPIncidents import fetch_incidents

    http_mocker = ClientMock(100, num_of_alerts=100, num_of_incidents=100)
    mocker.patch.object(client, "_http_request", side_effect=http_mocker.http_request_side_effect)

    sinceDate = "2009-01-20T00:00:00Z"
    sinceDate = arg_to_datetime(sinceDate)
    last_run = {"incidents": {"last_fetch": 0}}
    next_run, events = fetch_incidents(100, last_run, True, ["high"], [], client, sinceDate)
    assert len(events) == 100


def test_search_find_command(mocker, client: Client):
    http_mocker = ClientMock(5000, num_of_alerts=2500, num_of_incidents=2500)
    mocker.patch.object(client, "_http_request", side_effect=http_mocker.http_request_side_effect)
    res = search_find(client, {"query": "md5"})
    assert len(res) is not None