Alibaba Action Trail Event Collector

import base64
import hashlib
import hmac
from datetime import datetime
from typing import Any

import demistomock as demisto
import six
import urllib3
from CommonServerPython import *
from SiemApiModule import *

API_VERSION = "0.6.0"
VENDOR = "alibaba"
PRODUCT = "action_trail"
urllib3.disable_warnings()


class AlibabaParams(BaseModel):
    from_: str = Field(alias="from")
    to: int
    type: str = "log"
    offset: int = 0
    reverse: bool = False
    powerSql: bool = False
    query: str


class AlibabaEventsClient(IntegrationEventsClient):
    def __init__(
        self,
        request: IntegrationHTTPRequest,
        options: IntegrationOptions,
        access_key: str,
        access_key_id: str,
        logstore_name: str,
    ):
        self.access_key = access_key
        self.access_key_id = access_key_id
        self.logstore_name = logstore_name
        super().__init__(request=request, options=options)

    def set_request_filter(self, after: Any):
        from_time = int(after)

        self.request.params.from_ = from_time + 1  # type: ignore
        self.request.params.to = from_time + 3600  # type: ignore

    def call(self, request: IntegrationHTTPRequest) -> requests.Response:
        try:
            response = self.session.request(**self.request.dict(by_alias=True))
            if response.status_code >= 400:
                demisto.debug(f"An error occurred - raw response is: {response.json()}")
            response.raise_for_status()
            return response
        except Exception as exc:
            msg = f"something went wrong with the http call {exc}"
            demisto.debug(msg)
            raise DemistoException(msg) from exc

    def prepare_request(self):
        headers = self.request.headers

        del headers["x-log-date"]
        headers["Date"] = datetime.utcnow().strftime("%a, %d %b %Y %H:%M:%S GMT")

        signature = get_request_authorization(
            f"/logstores/{self.logstore_name}",
            self.access_key,
            self.request.params.dict(by_alias=True),  # type: ignore
            headers,  # type: ignore
        )  # type: ignore

        headers["Authorization"] = "LOG " + self.access_key_id + ":" + signature
        headers["x-log-date"] = headers["Date"]
        del headers["Date"]

        self.request.headers = headers


class AlibabaGetEvents(IntegrationGetEvents):
    client: AlibabaEventsClient

    def __init__(self, client: AlibabaEventsClient, options: IntegrationOptions):
        super().__init__(client=client, options=options)

    @staticmethod
    def get_last_run(events: list) -> dict:
        return {"from": events[-1]["__time__"]}

    def _iter_events(self):
        self.client.prepare_request()
        response = self.call()
        events: list = response.json()
        events.sort(key=lambda k: k.get("__time__"))

        if not events:
            return []

        while True:
            yield events

            last = events[-1]
            self.client.set_request_filter(last["__time__"])
            self.client.prepare_request()
            response = self.call()

            events = response.json()
            events.sort(key=lambda k: k.get("__time__"))

            if not events:
                break


def canonicalized_log_headers(headers):
    content = ""
    for key in sorted(six.iterkeys(headers)):
        if key[:6].lower() in ("x-log-", "x-acs-"):  # x-log- header
            content += key + ":" + str(headers[key]) + "\n"
    return content


def canonicalized_resource(resource, params):
    if params:
        urlString = ""
        for key, value in sorted(six.iteritems(params)):
            urlString += "{}={}&".format(key, value.decode("utf8") if isinstance(value, six.binary_type) else value)
        resource += "?" + urlString[:-1]
    return resource


def base64_encodestring(s):
    if isinstance(s, str):
        s = s.encode("utf8")
    return base64.encodebytes(s).decode("utf8")


def hmac_sha1(content, key):
    if isinstance(content, six.text_type):  # hmac.new accept 8-bit str
        content = content.encode("utf-8")
    if isinstance(key, six.text_type):  # hmac.new accept 8-bit str
        key = key.encode("utf-8")

    hashed = hmac.new(key, content, hashlib.sha1).digest()
    return base64_encodestring(hashed).rstrip()


def get_request_authorization(resource, key, req_params, req_headers):
    content = "GET\n\n\n"
    content += req_headers["Date"] + "\n"
    content += canonicalized_log_headers(req_headers)
    content += canonicalized_resource(resource, req_params)
    return hmac_sha1(content, key)


def get_alibaba_timestamp_format(value):
    timestamp: datetime
    if isinstance(value, int):
        return value
    if not isinstance(value, datetime):
        timestamp = dateparser.parse(value)  # type: ignore
    return int(time.mktime(timestamp.timetuple()))


def main():
    # Args is always stronger. Get last run even stronger
    demisto_params = demisto.params() | demisto.args() | demisto.getLastRun()

    project_name = demisto_params.get("project_name")
    endpoint = demisto_params.get("endpoint")
    logstore_name = demisto_params.get("logstore_name")
    access_key = demisto_params.get("access_key").get("password")
    access_key_id = demisto_params.get("access_key").get("identifier")
    query = demisto_params.get("query")
    from_ = get_alibaba_timestamp_format(demisto_params.get("from") or "3 days")
    should_push_events = argToBoolean(demisto_params.get("should_push_events", "false"))

    headers = {
        "Content-Length": "0",
        "x-log-bodyrawsize": "0",
        "x-log-apiversion": API_VERSION,
        "x-log-signaturemethod": "hmac-sha1",
        "Host": f"{project_name}.{endpoint}",
        "x-log-date": "",
    }

    params = {"from": str(from_), "to": str(from_ + 3600), "query": query}

    demisto_params["method"] = Method.GET
    demisto_params["url"] = f"http://{project_name}.{endpoint}:80/logstores/{logstore_name}"
    demisto_params["headers"] = headers

    request = IntegrationHTTPRequest(**demisto_params)
    request.params = AlibabaParams.model_validate(params)  # type: ignore[attr-defined,assignment]

    options = IntegrationOptions.model_validate(demisto_params)  # type: ignore[attr-defined]

    client = AlibabaEventsClient(
        request, options, access_key=access_key, access_key_id=access_key_id, logstore_name=logstore_name
    )

    get_events = AlibabaGetEvents(client, options)

    command = demisto.command()
    try:
        if command == "test-module":
            get_events.client.options.limit = 1
            get_events.run()
            return_results("ok")
        elif command in ("alibaba-get-events", "fetch-events"):
            events = get_events.run()

            if command == "fetch-events":
                send_events_to_xsiam(events, vendor=VENDOR, product=PRODUCT)
                if events:
                    demisto.setLastRun(AlibabaGetEvents.get_last_run(events))

            elif command == "alibaba-get-events":
                command_results = CommandResults(
                    readable_output=tableToMarkdown("alibaba Logs", events, headerTransform=pascalToSpace),
                    outputs_prefix="alibaba.Logs",
                    outputs_key_field="event.eventid",
                    outputs=events,
                    raw_response=events,
                )
                return_results(command_results)

                if should_push_events:
                    send_events_to_xsiam(events, vendor=VENDOR, product=PRODUCT)

    except Exception as e:
        return_error(str(e))


if __name__ in ("__main__", "__builtin__", "builtins"):
    main()