Connection Event Logs

security_logs.py

from __future__ import annotations
import datetime as dt
import json

import httpx  # pip install httpx


class RESTAPIV1(httpx.Client):
    """It's just an HTTP client."""

    def session_login(self, username: str, password: str) -> httpx.Response:
        # https://developers.thoughtspot.com/docs/?pageid=session-api#session-login
        d = {"username": username, "password": password, "rememberme": True}
        return self.post("/callosum/v1/tspublic/v1/session/login", data=d)

    def logs(self, from_date: dt.date, to_date: dt.date = None, topic: str = "security_logs") -> httpx.Response:
        # https://developers.thoughtspot.com/docs/logs-api
        if to_date is None:
            to_date = from_date + dt.timedelta(hours=24)

        epoch_from_s = int(dt.datetime.combine(from_date, dt.time()).timestamp())
        epoch_to_s   = int(dt.datetime.combine(to_date, dt.time()).timestamp())

        p = {"topic": topic, "fromEpoch": epoch_from_s * 1000, "toEpoch": epoch_to_s * 1000}
        return self.get(f"/callosum/v1/tspublic/v1/logs/topics/{topic}", params=p)

    def session_logout(self) -> httpx.Response:
        # https://developers.thoughtspot.com/docs/?pageid=session-api#session-logout
        return self.post("/callosum/v1/tspublic/v1/session/logout")


def main():
    """
    Object is to get the parent LOGICAL_TABLE for a specific Filter chip on a Liveboard.
    """
    TS_URL = "..."
    TS_USERNAME = "..."
    TS_PASSWORD = "..."

    client = RESTAPIV1(base_url=TS_URL, timeout=60 * 5)
    r = client.session_login(username=TS_USERNAME, password=TS_PASSWORD)
    print(f"session/login << {r.status_code}")

    if r.status_code != 204:
        print(r.text)
        return

    events = []
    events_we_care_about = {"CREATE_CONNECTION", "EDIT_CONNECTION", "DELETE_CONNECTION"}

    # Fetch information from the audit logs
    end = dt.datetime.combine(dt.datetime.today().date(), dt.time())

    while True:
        start = end - dt.timedelta(hours=6)

        p = {"topic": "security_logs", "fromEpoch": int(start.timestamp()) * 1000, "toEpoch": int(end.timestamp()) * 1000}
        r = client.get("/callosum/v1/tspublic/v1/logs/topics/security_logs", params=p)

        logs = [{**entry, "log": json.loads(entry["log"])} for entry in r.json()]
        cnxn = [entry for entry in logs if entry["log"]["type"] in events_we_care_about]
        events.extend(logs)

        print(
            f"logs/topics/security_logs << {r.status_code}"
            f"\n --> fetched: {len(logs): >6,} events, ({len(cnxn): >3} connection events)"
        )

        if (dt.datetime.today() - start) >= dt.timedelta(days=5):
            break

        end = start

    # print(json.dumps(events, indent=4))

    r = client.session_logout()
    print(f"session/logout << {r.status_code}")


if __name__ == "__main__":
    raise SystemExit(main())