#!/usr/bin/env python3
"""Honeylog API from the command line.
A thin wrapper over the Honeylog REST API v1, the same API the MCP server
forwards to. Every command is one endpoint, and one MCP tool, and every request
reads a single site, custom view or network:
honeylog.py sites --search example.com
honeylog.py fields site:ab12cd --source raw_logs
honeylog.py timeseries site:ab12cd --metrics views,sessions --range last_30_days --compare-range previous_period
honeylog.py breakdown site:ab12cd --dimension bot_name --range yesterday --with-totals --limit 50
honeylog.py raw-logs site:ab12cd --last 2h --columns datetime_second,bot_name,full_url --all
honeylog.py raw-logs-timeseries site:ab12cd --range today --granularity 15m
Authentication: run "honeylog.py login" once. It opens Honeylog in the browser,
you allow access, and the token it gets is saved in credentials.json in the
user's config directory, so every new terminal is already signed in until the
token expires (60 days unless the server says otherwise). "honeylog.py logout"
revokes it and removes it. A team API token from Settings -> API Tokens, MCP &
CLI in HONEYLOG_TOKEN takes precedence over the login, for scripts and CI (a
--token flag also works, but other users can see it in the process list).
The base URL is --base-url, else HONEYLOG_BASE_URL, else the one of the last
login, else https://honeylog.io.
Documentation: https://honeylog.io/docs/cli
Output: a table on a terminal, JSON otherwise; --format table|json|jsonl|csv
to choose. Rate-limited requests are retried after the delay the API asks for.
Requires Python 3.8 or newer and nothing else.
"""
from __future__ import annotations
import argparse
import base64
import contextlib
import csv
import email.utils
import hashlib
import hmac
import html
import json
import os
import random
import re
import secrets
import socket
import sys
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
import webbrowser
from datetime import datetime, timedelta, timezone
from http.server import BaseHTTPRequestHandler, HTTPServer
from typing import Any, Callable, Dict, Iterable, Iterator, List, Optional, Sequence, TextIO, Tuple
VERSION = "1.2.0"
DEFAULT_BASE_URL = "https://honeylog.io"
# The name login registers with, followed by " on {hostname}". The server
# recognises the CLI by it (App\Models\McpClient::COMMAND_LINE_NAME, kept equal
# by tests/Unit/HoneylogCliTest.php): it only changes the words of the consent
# page and the name of the token.
CLIENT_NAME = "Honeylog CLI"
CREDENTIALS_FILE = "credentials.json"
CALLBACK_PATH = "/callback"
LOGIN_TIMEOUT_SECONDS = 600.0
# Below this, a command run in a terminal says that the login is about to expire.
EXPIRY_WARNING = timedelta(days=7)
# The grammar the API routes accept for {wireable}: kept identical to
# ChecksApiAccess::WIREABLE_PATTERN (tests/Unit/HoneylogCliTest.php checks it).
WIREABLE_PATTERN = r"^(?:site:[0-9a-zA-Z\-]{1,64}(?:@[0-9]+)?|network:[0-9]+)$"
WIREABLE_RE = re.compile(WIREABLE_PATTERN)
# The API route behind each command, after /api/v1/ and the wireable. Every MCP
# tool has a command here with the same arguments as flags, and the same
# allowed values as below (tests/Feature/Mcp/McpCliParityTest.php checks it).
COMMAND_ENDPOINTS = {
"sites": "sites",
"fields": "fields",
"timeseries": "stats/views",
"breakdown": "stats/pages",
"raw-logs": "raw-logs",
"raw-logs-timeseries": "raw-logs/timeseries",
}
STATS_RANGES = ("yesterday", "last_7_days", "last_14_days", "last_30_days", "last_90_days", "last_365_days", "this_month", "last_month", "this_year", "last_year")
STATS_RANGE_HELP = "whole days, ending yesterday at the latest"
COMPARE_RANGES = ("previous_period", "previous_month", "previous_year")
RAW_LOGS_RANGES = ("today", "yesterday", "last_7_days")
RAW_LOGS_RANGE_HELP = "today and last_7_days run up to now"
RAW_LOGS_GRANULARITIES = ("1m", "5m", "15m", "1h", "1d")
EXIT_OK = 0
EXIT_UNEXPECTED = 1
EXIT_USAGE = 2
EXIT_AUTH = 3
EXIT_NOT_FOUND = 4
EXIT_INVALID = 5
EXIT_RATE_LIMITED = 6
EXIT_SERVER = 7
EXIT_NETWORK = 8
EXIT_INTERRUPTED = 130
EXIT_CODES_HELP = """exit codes:
0 ok
1 unexpected error
2 wrong usage, or no token at all
3 token rejected (HTTP 401/403), login expired, or the login did not complete
4 site group not found (HTTP 404)
5 invalid request (HTTP 422): the errors are printed per field
6 still rate limited after the retries (HTTP 429)
7 server error (HTTP 5xx, including 504 for a query that was too expensive)
8 network error
"""
RETRYABLE_STATUSES = (429, 503)
MAX_BACKOFF_SECONDS = 60.0
MAX_TOTAL_WAIT_SECONDS = 300.0
TABLE_CELL_WIDTH = 60
DEFAULT_PAGE_SIZE = 100
LOOPBACK_HOSTS = ("localhost", "127.0.0.1", "::1")
class UsageError(Exception):
"""A problem with the command line itself: nothing was sent."""
class NetworkError(Exception):
"""The API could not be reached, even after the retries."""
class LoginError(Exception):
"""The browser login did not give a token, or the saved one can no longer be used."""
class ApiError(Exception):
"""The API answered with an error status."""
def __init__(self, status: int, message: str, errors: Optional[Dict[str, Any]] = None):
super().__init__(message)
self.status = status
self.message = message
self.errors = errors or {}
@property
def exit_code(self) -> int:
if self.status in (401, 403):
return EXIT_AUTH
if self.status == 404:
return EXIT_NOT_FOUND
if self.status == 422:
return EXIT_INVALID
if self.status == 429:
return EXIT_RATE_LIMITED
if self.status >= 500:
return EXIT_SERVER
return EXIT_UNEXPECTED
class Client:
"""GET requests to /api/v1 with the bearer token, retrying on 429/503 and network errors."""
def __init__(
self,
base_url: str,
token: str,
timeout: float = 60.0,
retries: int = 3,
verbose: bool = False,
stderr: Optional[TextIO] = None,
sleep: Callable[[float], None] = time.sleep,
):
self.base_url = base_url.rstrip("/")
self.token = token
self.timeout = timeout
self.retries = max(0, retries)
self.verbose = verbose
self.stderr = stderr or sys.stderr
self.sleep = sleep
def get(self, path: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
query = {key: value for key, value in (params or {}).items() if value is not None and value != ""}
url = self.base_url + "/api/v1/" + path.lstrip("/")
if query:
url += "?" + urllib.parse.urlencode(query)
waited = 0.0
attempt = 0
while True:
if self.verbose:
self.stderr.write("GET {}\n".format(url))
try:
status, headers, payload = self._request(url)
except NetworkError:
delay = self._backoff(attempt)
if attempt >= self.retries or waited + delay > MAX_TOTAL_WAIT_SECONDS:
raise
else:
if 200 <= status < 300:
return payload
message = payload.get("message") if isinstance(payload.get("message"), str) else "HTTP {}".format(status)
errors = payload.get("errors") if isinstance(payload.get("errors"), dict) else None
if status not in RETRYABLE_STATUSES:
raise ApiError(status, message, errors)
delay = self._retry_delay(attempt, headers.get("Retry-After"), message)
if attempt >= self.retries or waited + delay > MAX_TOTAL_WAIT_SECONDS:
raise ApiError(status, message, errors)
if self.verbose:
self.stderr.write("retrying in {:.1f}s\n".format(delay))
self.sleep(delay)
waited += delay
attempt += 1
def _request(self, url: str):
request = urllib.request.Request(url, method="GET", headers={
"Authorization": "Bearer " + self.token,
"Accept": "application/json",
"User-Agent": "honeylog-cli/" + VERSION,
})
try:
with urllib.request.urlopen(request, timeout=self.timeout) as response:
return response.status, response.headers, _decode_json(response.read())
except urllib.error.HTTPError as error:
return error.code, error.headers, _decode_json(error.read())
except (urllib.error.URLError, socket.timeout, ConnectionError) as error:
reason = getattr(error, "reason", error)
raise NetworkError("cannot reach {}: {}".format(self.base_url, reason)) from error
def _retry_delay(self, attempt: int, retry_after: Optional[str], message: str) -> float:
if retry_after:
try:
return max(0.0, float(retry_after))
except ValueError:
try:
when = email.utils.parsedate_to_datetime(retry_after)
return max(0.0, (when - datetime.now(timezone.utc)).total_seconds())
except (TypeError, ValueError):
pass
# ApiRequestLimiter also says it in words: "Try again in 12 seconds."
match = re.search(r"in (\d+) seconds?", message)
if match:
return float(match.group(1))
return self._backoff(attempt)
@staticmethod
def _backoff(attempt: int) -> float:
return min(MAX_BACKOFF_SECONDS, 2.0 ** attempt) + random.uniform(0, 1)
def _decode_json(body: bytes) -> Dict[str, Any]:
try:
decoded = json.loads(body.decode("utf-8") or "{}")
except (UnicodeDecodeError, ValueError):
return {"message": body[:200].decode("utf-8", "replace").strip() or "invalid response"}
return decoded if isinstance(decoded, dict) else {"data": decoded}
def encode_expression(raw: str, stdin: TextIO) -> str:
"""A filter expression as JSON text, @file or - (stdin), encoded the way the dashboard links do."""
if raw == "-":
text = stdin.read()
elif raw.startswith("@"):
try:
with open(raw[1:], "r", encoding="utf-8") as handle:
text = handle.read()
except OSError as error:
raise UsageError("cannot read {}: {}".format(raw[1:], error.strerror))
else:
text = raw
try:
expression = json.loads(text)
except ValueError as error:
raise UsageError("the filter expression is not valid JSON: {}".format(error))
if not isinstance(expression, list):
raise UsageError('a filter expression is a JSON list, such as [[["pathname","=","/pricing"]]]')
# Same as ExpressionUrlCodec::encode(): base64url of the JSON, no padding.
encoded = base64.urlsafe_b64encode(json.dumps(expression, separators=(",", ":")).encode("utf-8"))
return encoded.decode("ascii").rstrip("=")
def parse_duration(value: str) -> int:
match = re.fullmatch(r"\s*(\d+)\s*([smhd])\s*", value or "")
if not match:
raise UsageError("--last takes a number and a unit, such as 30m, 6h or 2d")
return int(match.group(1)) * {"s": 1, "m": 60, "h": 3600, "d": 86400}[match.group(2)]
def check_range(args: argparse.Namespace) -> None:
if args.range and (args.start or args.end):
raise UsageError("--range names the whole window: use it instead of --start and --end")
def resolve_raw_window(args: argparse.Namespace) -> Tuple[Any, Any]:
"""The start and end a raw-logs request sends: as given, or counted back from now with --last."""
check_range(args)
if not args.last:
return args.start, args.end
if args.start or args.end or args.range:
raise UsageError("--last counts back from now: use it instead of --start, --end and --range")
end = int(time.time())
return end - parse_duration(args.last), end
def check_wireable(value: str) -> str:
if not WIREABLE_RE.match(value):
raise UsageError(
'a site group looks like "site:{uuid}", "site:{uuid}@{customViewId}" or "network:{id}": '
"run the sites command to get the ones your token can read"
)
return value
def split_list(value: Optional[str]) -> Optional[str]:
if value is None:
return None
return ",".join(part.strip() for part in value.split(",") if part.strip()) or None
# ---------------------------------------------------------------------------
# Output
# ---------------------------------------------------------------------------
def format_cell(value: Any) -> str:
if value is None:
return ""
if isinstance(value, bool):
return "true" if value else "false"
if isinstance(value, (dict, list)):
return json.dumps(value, separators=(",", ":"), ensure_ascii=False)
return str(value)
def columns_of(rows: Sequence[Dict[str, Any]]) -> List[str]:
columns: List[str] = []
for row in rows:
for key in row:
if key not in columns:
columns.append(key)
return columns
def render_table(rows: Sequence[Dict[str, Any]], out: TextIO) -> None:
if not rows:
out.write("(no rows)\n")
return
columns = columns_of(rows)
cells = [[_truncate(format_cell(row.get(column))) for column in columns] for row in rows]
widths = [max([len(column)] + [len(line[index]) for line in cells]) for index, column in enumerate(columns)]
out.write(" ".join(column.ljust(widths[index]) for index, column in enumerate(columns)).rstrip() + "\n")
out.write(" ".join("-" * width for width in widths) + "\n")
for line in cells:
out.write(" ".join(cell.ljust(widths[index]) for index, cell in enumerate(line)).rstrip() + "\n")
def _truncate(text: str) -> str:
text = text.replace("\n", " ")
return text if len(text) <= TABLE_CELL_WIDTH else text[: TABLE_CELL_WIDTH - 1] + "…"
class RowWriter:
"""Writes pages of rows: csv and jsonl as they arrive, table and json at the end."""
def __init__(self, fmt: str, out: TextIO):
self.fmt = fmt
self.out = out
self.buffered: List[Dict[str, Any]] = []
self.csv_writer: Optional[csv.DictWriter] = None
def write(self, rows: Iterable[Dict[str, Any]]) -> None:
rows = list(rows)
if self.fmt == "jsonl":
for row in rows:
self.out.write(json.dumps(row, ensure_ascii=False) + "\n")
elif self.fmt == "csv":
if self.csv_writer is None and rows:
self.csv_writer = csv.DictWriter(self.out, fieldnames=columns_of(rows), extrasaction="ignore", restval="")
self.csv_writer.writeheader()
for row in rows:
self.csv_writer.writerow({key: format_cell(value) for key, value in row.items()})
else:
self.buffered.extend(rows)
def finish(self, meta: Optional[Dict[str, Any]] = None) -> None:
if self.fmt == "table":
render_table(self.buffered, self.out)
elif self.fmt == "json":
self.out.write(json.dumps({"data": self.buffered, "meta": meta or {}}, indent=2, ensure_ascii=False) + "\n")
def print_json(payload: Dict[str, Any], out: TextIO) -> None:
out.write(json.dumps(payload, indent=2, ensure_ascii=False) + "\n")
# ---------------------------------------------------------------------------
# Commands
# ---------------------------------------------------------------------------
def site_rows(payload: Dict[str, Any]) -> List[Dict[str, Any]]:
data = payload.get("data") or {}
rows: List[Dict[str, Any]] = []
for site in data.get("sites") or []:
team = (site.get("team") or {}).get("name")
rows.append(_with_team({"type": "site", "wireable": site.get("wireable"), "name": site.get("domain"), "timezone": site.get("timezone")}, team, site))
for view in site.get("custom_views") or []:
rows.append(_with_team({"type": "view", "wireable": view.get("wireable"), "name": "{} / {}".format(site.get("domain"), view.get("name")), "timezone": site.get("timezone")}, team, site))
for network in data.get("networks") or []:
rows.append(_with_team({"type": "network", "wireable": network.get("wireable"), "name": network.get("name"), "timezone": ""}, (network.get("team") or {}).get("name"), network))
return rows
def _with_team(row: Dict[str, Any], team: Optional[str], item: Dict[str, Any]) -> Dict[str, Any]:
# Only a token that reads more than one team gets the team of every entry.
if "team" in item:
row["team"] = team
return row
def command_sites(client: Client, args: argparse.Namespace, out: TextIO, warn: Callable[[str], None]) -> None:
params: Dict[str, Any] = {
"search": args.search,
"type": args.type,
"team_id": args.team_id,
"limit": args.limit if args.limit is not None else (DEFAULT_PAGE_SIZE if args.all else None),
"offset": args.offset,
}
page = client.get(COMMAND_ENDPOINTS["sites"], params)
if not args.all:
if args.format == "json":
print_json(page, out)
return
writer = RowWriter(args.format, out)
writer.write(site_rows(page))
writer.finish()
if (page.get("meta") or {}).get("has_more"):
warn("more entries are available: add --all, or narrow the list with --search")
return
sites = list((page.get("data") or {}).get("sites") or [])
networks = list((page.get("data") or {}).get("networks") or [])
writer = RowWriter(args.format, out)
writer.write(site_rows(page))
meta = page.get("meta") or {}
while meta.get("has_more") and meta.get("next_offset") is not None and len(sites) + len(networks) < args.max_rows:
params["offset"] = meta["next_offset"]
page = client.get(COMMAND_ENDPOINTS["sites"], params)
sites.extend((page.get("data") or {}).get("sites") or [])
networks.extend((page.get("data") or {}).get("networks") or [])
writer.write(site_rows(page))
meta = page.get("meta") or {}
if meta.get("has_more"):
warn("stopped after {} entries (--max-rows): narrow the list with --search".format(len(sites) + len(networks)))
if args.format == "json":
print_json({"data": {"sites": sites, "networks": networks}, "meta": {"sites": len(sites), "networks": len(networks)}}, out)
else:
writer.finish()
def command_fields(client: Client, args: argparse.Namespace, out: TextIO, warn: Callable[[str], None]) -> None:
page = client.get(check_wireable(args.wireable) + "/" + COMMAND_ENDPOINTS["fields"], {"source": args.source})
if args.format == "json":
print_json(page, out)
return
data = page.get("data") or {}
rows: List[Dict[str, Any]] = []
# In the raw-logs dictionary every dimension is a column you can ask for.
dimension_kind = "column" if args.source == "raw_logs" else "dimension"
for kind, key in (("metric", "metrics"), (dimension_kind, "dimensions")):
for field in data.get(key) or []:
rows.append({
"kind": kind,
"name": field.get("name"),
"display_name": field.get("display_name"),
"type": field.get("type"),
"description": field.get("description"),
})
writer = RowWriter(args.format, out)
writer.write(rows)
writer.finish()
def command_timeseries(client: Client, args: argparse.Namespace, out: TextIO, warn: Callable[[str], None], stdin: TextIO) -> None:
wireable = check_wireable(args.wireable)
check_range(args)
params = {
"range": args.range,
"start": args.start,
"end": args.end,
"compare_range": args.compare_range,
"compare_start": args.compare_start,
"compare_end": args.compare_end,
"granularity": args.granularity,
"metrics": split_list(args.metrics),
"where_expression": encode_expression(args.where, stdin) if args.where else None,
"having_expression": encode_expression(args.having, stdin) if args.having else None,
}
page = client.get(wireable + "/" + COMMAND_ENDPOINTS["timeseries"], params)
if args.format == "json":
print_json(page, out)
return
data = page.get("data") or {}
labels = data.get("labels") or []
compare_labels = data.get("compare_labels") or []
rows = []
for index, label in enumerate(labels):
row: Dict[str, Any] = {"label": label}
if compare_labels:
row["compare_label"] = compare_labels[index] if index < len(compare_labels) else None
for series in data.get("series") or []:
values = series.get("data") or []
row[series.get("label")] = values[index] if index < len(values) else None
rows.append(row)
writer = RowWriter(args.format, out)
writer.write(rows)
writer.finish()
def command_breakdown(client: Client, args: argparse.Namespace, out: TextIO, warn: Callable[[str], None], stdin: TextIO) -> None:
path = check_wireable(args.wireable) + "/" + COMMAND_ENDPOINTS["breakdown"]
check_range(args)
params: Dict[str, Any] = {
"range": args.range,
"start": args.start,
"end": args.end,
"compare_range": args.compare_range,
"compare_start": args.compare_start,
"compare_end": args.compare_end,
"dimension": args.dimension,
"dimensions": split_list(args.dimensions),
"metrics": split_list(args.metrics),
"sort": args.sort,
"sort_order": args.sort_order,
"where_expression": encode_expression(args.where, stdin) if args.where else None,
"having_expression": encode_expression(args.having, stdin) if args.having else None,
"with_totals": "true" if args.with_totals else None,
"limit": args.limit,
"offset": args.offset,
}
page = client.get(path, params)
if not args.all and args.format == "json":
print_json(page, out)
return
writer = RowWriter(args.format, out)
rows = page.get("data") or []
writer.write(rows)
meta = page.get("meta") or {}
seen = len(rows)
if args.all:
# This endpoint has no has_more: a full page means there may be another.
# The window is pinned to the first page, or "now" would move it; the
# API refuses range next to start and end, and the totals are already in.
params["start"], params["end"], params["range"] = meta.get("start"), meta.get("end"), None
params["with_totals"] = None
limit = int(meta.get("limit") or len(rows) or 1)
offset = int(meta.get("offset") or 0)
while len(rows) == limit and seen < args.max_rows:
offset += limit
params["offset"] = offset
page = client.get(path, params)
rows = page.get("data") or []
writer.write(rows)
seen += len(rows)
if len(rows) == limit and seen >= args.max_rows:
warn("stopped after {} rows (--max-rows): more rows may be available".format(seen))
writer.finish(dict(meta, rows=seen))
if args.format == "table" and isinstance(meta.get("totals"), dict):
out.write("\ntotals over every matching row\n")
render_table([meta["totals"]], out)
def command_raw_logs(client: Client, args: argparse.Namespace, out: TextIO, warn: Callable[[str], None], stdin: TextIO) -> None:
wireable = check_wireable(args.wireable)
path = wireable + "/" + COMMAND_ENDPOINTS["raw-logs"]
start, end = resolve_raw_window(args)
limit = args.limit
if args.all and limit is None:
limit = min(discover_raw_logs_page_size(client, wireable), args.max_rows)
params: Dict[str, Any] = {
"range": args.range,
"start": start,
"end": end,
"columns": split_list(args.columns),
"where_expression": encode_expression(args.where, stdin) if args.where else None,
"order_direction": args.order,
"limit": limit,
"offset": args.offset,
}
page = client.get(path, params)
if not args.all and args.format == "json":
print_json(page, out)
return
writer = RowWriter(args.format, out)
rows = page.get("data") or []
writer.write(rows)
meta = page.get("meta") or {}
first_meta = dict(meta)
seen = len(rows)
if args.all:
# Pin the window to the one the first page resolved: with the default
# end ("now"), or a range such as today, every page would otherwise
# look at a different window. The API refuses range next to start and end.
params["start"], params["end"], params["range"] = meta.get("start"), meta.get("end"), None
while meta.get("has_more") and meta.get("next_offset") is not None and seen < args.max_rows:
params["offset"] = meta["next_offset"]
params["limit"] = min(int(meta.get("limit") or DEFAULT_PAGE_SIZE), args.max_rows - seen)
page = client.get(path, params)
rows = page.get("data") or []
writer.write(rows)
seen += len(rows)
meta = page.get("meta") or {}
if meta.get("has_more"):
if meta.get("next_offset") is None:
warn("stopped at the API offset limit after {} rows: narrow --start/--end or add --where to reach the rest".format(seen))
else:
warn("stopped after {} rows (--max-rows): more rows are available".format(seen))
elif meta.get("has_more"):
warn("more rows are available: add --all, or narrow the window")
writer.finish(dict(first_meta, rows=seen, has_more=bool(meta.get("has_more"))))
def command_raw_logs_timeseries(client: Client, args: argparse.Namespace, out: TextIO, warn: Callable[[str], None], stdin: TextIO) -> None:
wireable = check_wireable(args.wireable)
start, end = resolve_raw_window(args)
params = {
"range": args.range,
"start": start,
"end": end,
"granularity": args.granularity,
"where_expression": encode_expression(args.where, stdin) if args.where else None,
}
page = client.get(wireable + "/" + COMMAND_ENDPOINTS["raw-logs-timeseries"], params)
if args.format == "json":
print_json(page, out)
return
writer = RowWriter(args.format, out)
writer.write(page.get("data") or [])
writer.finish()
def discover_raw_logs_page_size(client: Client, wireable: str) -> int:
"""The largest page the server accepts, from the limits it publishes (one extra call)."""
try:
limits = (client.get(wireable + "/" + COMMAND_ENDPOINTS["fields"], {"source": "raw_logs"}).get("meta") or {}).get("limits") or {}
except ApiError:
return DEFAULT_PAGE_SIZE
max_limit = limits.get("max_limit")
return int(max_limit) if isinstance(max_limit, int) and max_limit > 0 else DEFAULT_PAGE_SIZE
# ---------------------------------------------------------------------------
# Login
# ---------------------------------------------------------------------------
def config_dir(env: Dict[str, str]) -> str:
"""Where the login is saved: HONEYLOG_CONFIG_DIR, else the user's own config directory."""
if env.get("HONEYLOG_CONFIG_DIR"):
return env["HONEYLOG_CONFIG_DIR"]
if os.name == "nt" and env.get("APPDATA"):
return os.path.join(env["APPDATA"], "honeylog")
return os.path.join(env.get("XDG_CONFIG_HOME") or os.path.join(os.path.expanduser("~"), ".config"), "honeylog")
def host_key(base_url: str) -> str:
"""The base URL a login is saved under: https://Honeylog.io/ and https://honeylog.io are one login."""
parts = urllib.parse.urlsplit(base_url.strip())
return urllib.parse.urlunsplit((parts.scheme.lower(), parts.netloc.lower(), parts.path.rstrip("/"), "", ""))
def _utc_now() -> datetime:
return datetime.now(timezone.utc)
def _parse_time(value: Any) -> Optional[datetime]:
if not isinstance(value, str):
return None
try:
parsed = datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ")
except ValueError:
return None
return parsed.replace(tzinfo=timezone.utc)
def _format_time(value: datetime) -> str:
return value.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
class CredentialStore:
"""The tokens login saved, one per Honeylog address, in a file only the user can read.
The file is read only when a command needs it: with --token or HONEYLOG_TOKEN
and --base-url or HONEYLOG_BASE_URL it is never opened.
"""
def __init__(self, path: str, warn: Callable[[str], None]):
self.path = path
self.warn = warn
self._data: Optional[Dict[str, Any]] = None
def _load(self) -> Dict[str, Any]:
if self._data is None:
data: Any = {}
try:
with open(self.path, "r", encoding="utf-8") as handle:
data = json.load(handle)
except FileNotFoundError:
pass
except (OSError, ValueError) as error:
self.warn("ignoring the saved login in {}: {}".format(self.path, getattr(error, "strerror", None) or error))
data = data if isinstance(data, dict) else {}
data["hosts"] = data.get("hosts") if isinstance(data.get("hosts"), dict) else {}
self._data = data
return self._data
def default_base_url(self) -> Optional[str]:
"""The address of the last login, used when neither --base-url nor HONEYLOG_BASE_URL says one."""
data = self._load()
default = data.get("default")
return default if isinstance(default, str) and default in data["hosts"] else None
def login_for(self, base_url: str) -> Optional[Dict[str, Any]]:
entry = self._load()["hosts"].get(host_key(base_url))
if not isinstance(entry, dict) or not isinstance(entry.get("token"), str) or not entry["token"]:
return None
return entry
def token_for(self, base_url: str) -> Optional[str]:
entry = self.login_for(base_url)
return entry["token"] if entry else None
def expires_at(self, base_url: str) -> Optional[datetime]:
entry = self.login_for(base_url)
return _parse_time(entry.get("expires_at")) if entry else None
def save_login(self, base_url: str, token: str, client_id: str, expires_in: Optional[int]) -> Optional[datetime]:
"""Saves the token and makes its address the default one; returns when it expires."""
data = self._load()
key = host_key(base_url)
now = _utc_now()
expires_at = now + timedelta(seconds=expires_in) if expires_in is not None else None
data["hosts"][key] = {
"token": token,
"client_id": client_id,
"saved_at": _format_time(now),
"expires_at": _format_time(expires_at) if expires_at else None,
}
data["default"] = key
self._write(data)
return expires_at
def remove(self, base_url: str) -> None:
data = self._load()
key = host_key(base_url)
data["hosts"].pop(key, None)
if data.get("default") == key:
data.pop("default", None)
if data["hosts"]:
self._write(data)
return
with contextlib.suppress(FileNotFoundError):
os.remove(self.path)
def _write(self, data: Dict[str, Any]) -> None:
directory = os.path.dirname(self.path) or "."
temporary = "{}.{}.tmp".format(self.path, os.getpid())
try:
os.makedirs(directory, mode=0o700, exist_ok=True)
# Created as 0600 and then moved in place: the token is never in a
# file that someone else could read, not even for a moment.
descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
json.dump(data, handle, indent=2)
handle.write("\n")
if os.name != "nt":
os.chmod(temporary, 0o600)
os.replace(temporary, self.path)
except OSError as error:
with contextlib.suppress(OSError):
os.remove(temporary)
raise UsageError("cannot save the login in {}: {}".format(self.path, error.strerror or error))
def http_json(method: str, url: str, timeout: float, form: Optional[Dict[str, str]] = None, body: Optional[Dict[str, Any]] = None) -> Tuple[int, Dict[str, Any]]:
"""One request to the sign-in endpoints, which live outside /api/v1 and are never retried."""
headers = {"Accept": "application/json", "User-Agent": "honeylog-cli/" + VERSION}
data = None
if form is not None:
data = urllib.parse.urlencode(form).encode("utf-8")
headers["Content-Type"] = "application/x-www-form-urlencoded"
elif body is not None:
data = json.dumps(body).encode("utf-8")
headers["Content-Type"] = "application/json"
request = urllib.request.Request(url, data=data, method=method, headers=headers)
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
return response.status, _decode_json(response.read())
except urllib.error.HTTPError as error:
return error.code, _decode_json(error.read())
except (urllib.error.URLError, socket.timeout, ConnectionError) as error:
parts = urllib.parse.urlsplit(url)
raise NetworkError("cannot reach {}://{}: {}".format(parts.scheme, parts.netloc, getattr(error, "reason", error))) from error
def _oauth_error(status: int, payload: Dict[str, Any]) -> str:
detail = payload.get("error_description") or payload.get("message") or payload.get("error")
return "HTTP {}{}".format(status, ": " + detail if isinstance(detail, str) and detail else "")
def check_endpoint(url: Any, allow_http: bool) -> str:
"""An address from the sign-in metadata gets a code or a token: same rule as --base-url."""
parts = urllib.parse.urlsplit(url if isinstance(url, str) else "")
if parts.hostname and (parts.scheme == "https" or (parts.scheme == "http" and (parts.hostname in LOOPBACK_HOSTS or allow_http))):
return url
raise LoginError("the sign-in metadata points to an address the CLI will not use: {!r}".format(url))
def discover_authorization_server(base_url: str, timeout: float, allow_http: bool) -> Dict[str, Optional[str]]:
"""The endpoints of the OAuth server (RFC 8414): the same one the MCP connectors sign in with."""
status, payload = http_json("GET", base_url.rstrip("/") + "/.well-known/oauth-authorization-server", timeout)
if not 200 <= status < 300:
raise LoginError(
"{} does not offer sign-in from the CLI ({}): create an API token in Settings -> API Tokens, MCP & CLI "
"and put it in HONEYLOG_TOKEN".format(base_url, _oauth_error(status, payload))
)
if "S256" not in (payload.get("code_challenge_methods_supported") or []):
raise LoginError("the sign-in service of {} does not support PKCE with S256".format(base_url))
scopes = payload.get("scopes_supported")
revocation = payload.get("revocation_endpoint")
return {
"authorization_endpoint": check_endpoint(payload.get("authorization_endpoint"), allow_http),
"token_endpoint": check_endpoint(payload.get("token_endpoint"), allow_http),
"registration_endpoint": check_endpoint(payload.get("registration_endpoint"), allow_http),
"revocation_endpoint": check_endpoint(revocation, allow_http) if revocation else None,
"scope": scopes[0] if isinstance(scopes, list) and scopes and isinstance(scopes[0], str) else None,
}
def client_name() -> str:
"""CLIENT_NAME and the machine, which is what the consent page and the token list show."""
host = re.sub(r"[^\w.\-]", "", socket.gethostname() or "")[:64]
return CLIENT_NAME + (" on " + host if host else "")
def _base64url(raw: bytes) -> str:
return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")
def pkce_pair() -> Tuple[str, str]:
"""A code verifier and its S256 challenge (RFC 7636)."""
verifier = _base64url(os.urandom(32))
return verifier, _base64url(hashlib.sha256(verifier.encode("ascii")).digest())
CALLBACK_PAGE = """
Honeylog CLI
{title}
{text}
"""
class LoginCallback:
"""The loopback address the browser comes back to (RFC 8252), and the answer it brings.
The answer can also be pasted in the terminal, when the browser runs on
another machine and cannot reach this one: accept() takes both.
"""
def __init__(self, state: str):
self.state = state
self.answer: Optional[Dict[str, str]] = None
self.received = threading.Event()
self._lock = threading.Lock()
self._serving = False
callback = self
class Handler(BaseHTTPRequestHandler):
def do_GET(self) -> None: # noqa: N802 - http.server naming
parts = urllib.parse.urlsplit(self.path)
if parts.path != CALLBACK_PATH:
self._reply(404, "Not found", "This address only receives the Honeylog sign-in.")
return
if not callback.accept(parts.query):
self._reply(400, "Not a sign-in answer", "This address has neither a code nor an error.")
elif callback.answer and "code" in callback.answer:
self._reply(200, "You are signed in", "The Honeylog CLI has what it needs: close this tab and go back to the terminal.")
else:
self._reply(400, "Sign-in not completed", "Go back to the terminal to see why, and run honeylog login again.")
def _reply(self, status: int, title: str, text: str) -> None:
page = CALLBACK_PAGE.format(title=html.escape(title), text=html.escape(text)).encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(page)))
self.send_header("Cache-Control", "no-store")
self.end_headers()
self.wfile.write(page)
def log_message(self, *args: Any) -> None:
pass
self.server = HTTPServer(("127.0.0.1", 0), Handler)
@property
def redirect_uri(self) -> str:
return "http://127.0.0.1:{}{}".format(self.server.server_address[1], CALLBACK_PATH)
def start(self) -> None:
threading.Thread(target=self.server.serve_forever, kwargs={"poll_interval": 0.1}, daemon=True).start()
self._serving = True
def stop(self) -> None:
if self._serving:
self.server.shutdown()
self._serving = False
self.server.server_close()
def accept(self, query: str) -> bool:
"""Takes the first answer that carries a code or an error; False if the query is neither."""
params = dict(urllib.parse.parse_qsl(query))
if "code" not in params and "error" not in params:
return False
# The state must be the one this login sent: anything else did not
# come from the consent page we opened.
if not hmac.compare_digest(params.get("state", ""), self.state):
params = {"error": "state_mismatch"}
with self._lock:
if self.answer is None:
self.answer = params
self.received.set()
return True
def wait(self, timeout: float) -> Optional[Dict[str, str]]:
# Short waits in a loop, so that Ctrl-C works on Windows as well.
deadline = time.monotonic() + timeout
while not self.received.wait(max(0.0, min(0.5, deadline - time.monotonic()))):
if time.monotonic() >= deadline:
return None
return self.answer
def read_pasted_answers(stdin: TextIO, callback: LoginCallback, warn: Callable[[str], None]) -> None:
for line in iter(stdin.readline, ""):
text = line.strip()
if not text:
continue
if callback.accept(urllib.parse.urlsplit(text).query if "?" in text else text):
return
warn("that has no code: paste the whole address from the address bar of the browser")
def can_open_browser(env: Dict[str, str], platform: Optional[str] = None) -> bool:
"""False where a browser would open out of sight: over SSH, or on a Unix without a display."""
platform = platform or sys.platform
if env.get("SSH_CONNECTION") or env.get("SSH_TTY"):
return False
if platform.startswith(("linux", "freebsd", "openbsd", "netbsd")) and not (env.get("DISPLAY") or env.get("WAYLAND_DISPLAY")):
return False
return True
def revoke_token(endpoints: Dict[str, Optional[str]], token: str, client_id: Optional[str], timeout: float) -> None:
endpoint = endpoints.get("revocation_endpoint")
if not endpoint:
raise LoginError("this Honeylog does not revoke tokens from the CLI")
form = {"token": token, "token_type_hint": "access_token"}
if client_id:
form["client_id"] = client_id
status, payload = http_json("POST", endpoint, timeout, form=form)
if not 200 <= status < 300:
raise LoginError(_oauth_error(status, payload))
def command_login(
args: argparse.Namespace,
base_url: str,
store: CredentialStore,
env: Dict[str, str],
stdin: TextIO,
stderr: TextIO,
warn: Callable[[str], None],
open_browser: Callable[[str], bool],
) -> None:
"""Authorization code with PKCE through the browser, then the token saved for the next sessions."""
endpoints = discover_authorization_server(base_url, args.timeout, args.allow_http)
verifier, challenge = pkce_pair()
callback = LoginCallback(secrets.token_urlsafe(24))
try:
callback.start()
status, registration = http_json("POST", str(endpoints["registration_endpoint"]), args.timeout, body={
"client_name": client_name(),
"redirect_uris": [callback.redirect_uri],
"grant_types": ["authorization_code"],
"response_types": ["code"],
"token_endpoint_auth_method": "none",
})
client_id = registration.get("client_id")
if not 200 <= status < 300 or not isinstance(client_id, str) or not client_id:
raise LoginError("Honeylog did not register the CLI ({})".format(_oauth_error(status, registration)))
query = {
"response_type": "code",
"client_id": client_id,
"redirect_uri": callback.redirect_uri,
"code_challenge": challenge,
"code_challenge_method": "S256",
"state": callback.state,
}
if endpoints["scope"]:
query["scope"] = str(endpoints["scope"])
authorization_endpoint = str(endpoints["authorization_endpoint"])
url = authorization_endpoint + ("&" if "?" in authorization_endpoint else "?") + urllib.parse.urlencode(query)
opened = False
if not args.no_browser and can_open_browser(env):
try:
opened = bool(open_browser(url))
except Exception: # noqa: BLE001 - any failure means: no browser here
opened = False
if opened:
warn("the sign-in page is open in your browser. If you don't see it, open this address:")
stderr.write("\n {}\n\n".format(url))
else:
warn("open this address in a browser, sign in and allow access:")
stderr.write("\n {}\n\n".format(url))
stderr.write(
"If the browser runs on another computer, it then ends on a 127.0.0.1 page that does not load:\n"
"copy the address of that page from the address bar and paste it here.\n"
)
threading.Thread(target=read_pasted_answers, args=(stdin, callback, warn), daemon=True).start()
warn("waiting for you to allow access in the browser (Ctrl-C to cancel)")
stderr.flush()
answer = callback.wait(LOGIN_TIMEOUT_SECONDS)
finally:
callback.stop()
if answer is None:
raise LoginError("no answer from the browser within {} minutes: run honeylog login again".format(int(LOGIN_TIMEOUT_SECONDS // 60)))
if answer.get("error") == "access_denied":
raise LoginError("access was not allowed in the browser")
if answer.get("error") == "state_mismatch":
raise LoginError("the browser came back with an answer that belongs to another login: run honeylog login again")
if "error" in answer:
raise LoginError("the sign-in failed: {}".format(answer.get("error_description") or answer["error"]))
status, payload = http_json("POST", str(endpoints["token_endpoint"]), args.timeout, form={
"grant_type": "authorization_code",
"code": answer.get("code", ""),
"redirect_uri": callback.redirect_uri,
"client_id": client_id,
"code_verifier": verifier,
})
token = payload.get("access_token")
if not 200 <= status < 300 or not isinstance(token, str) or not token:
raise LoginError("Honeylog did not issue a token ({})".format(_oauth_error(status, payload)))
expires_in = payload.get("expires_in")
previous = store.login_for(base_url)
try:
expires_at = store.save_login(base_url, token, client_id, int(expires_in) if isinstance(expires_in, (int, float)) else None)
except UsageError:
# A token nobody saved would only sit among the team's tokens.
with contextlib.suppress(LoginError, NetworkError):
revoke_token(endpoints, token, client_id, args.timeout)
raise
# A new login replaces the token saved before it, which nothing else uses:
# revoke it, or it would stay among the team's tokens until it expires.
if previous and previous["token"] != token:
try:
revoke_token(endpoints, previous["token"], previous.get("client_id"), args.timeout)
except (LoginError, NetworkError) as error:
warn("the token of the previous login could not be revoked ({}): delete it from Settings -> API Tokens, MCP & CLI".format(error))
until = " until {}".format(expires_at.strftime("%Y-%m-%d")) if expires_at else ""
warn("logged in to {}{}. The token is saved in {}, so new terminals are signed in too.".format(base_url, until, store.path))
if env.get("HONEYLOG_TOKEN"):
warn("HONEYLOG_TOKEN is set in this terminal and is used instead of the login: unset it to use the login")
def command_logout(args: argparse.Namespace, base_url: str, store: CredentialStore, env: Dict[str, str], warn: Callable[[str], None]) -> None:
login = store.login_for(base_url)
if login is None:
warn("not logged in to {}".format(base_url))
else:
try:
revoke_token(discover_authorization_server(base_url, args.timeout, args.allow_http), login["token"], login.get("client_id"), args.timeout)
problem = None
except (LoginError, NetworkError) as error:
problem = str(error)
store.remove(base_url)
if problem is None:
warn("logged out of {}: the token is revoked and removed from this computer".format(base_url))
else:
warn("logged out of {} on this computer, but the token could not be revoked ({}): delete it from Settings -> API Tokens, MCP & CLI".format(base_url, problem))
if env.get("HONEYLOG_TOKEN"):
warn("HONEYLOG_TOKEN is still set in this terminal, and the commands keep using it")
def resolve_token(args: argparse.Namespace, env: Dict[str, str], store: CredentialStore, base_url: str) -> Tuple[Optional[str], str]:
"""The token a command sends, and where it comes from: --token, HONEYLOG_TOKEN, then the login."""
if args.token:
return args.token, "--token"
if env.get("HONEYLOG_TOKEN"):
return env["HONEYLOG_TOKEN"], "HONEYLOG_TOKEN"
return store.token_for(base_url), "login"
def check_login_expiry(store: CredentialStore, base_url: str, stderr: TextIO, warn: Callable[[str], None]) -> None:
"""Refuses an expired login before any request, and says it is near only to a person at a terminal."""
expires_at = store.expires_at(base_url)
if expires_at is None:
return
left = expires_at - _utc_now()
if left.total_seconds() <= 0:
raise LoginError("the login to {} expired on {}: run honeylog login again".format(base_url, expires_at.strftime("%Y-%m-%d")))
if left < EXPIRY_WARNING and getattr(stderr, "isatty", lambda: False)():
warn("the login to {} expires on {}: run honeylog login to renew it".format(base_url, expires_at.strftime("%Y-%m-%d")))
# ---------------------------------------------------------------------------
# Command line
# ---------------------------------------------------------------------------
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="honeylog",
description="Read Honeylog analytics through the REST API. Every request reads one site, custom view or network. Start with: honeylog login",
epilog=EXIT_CODES_HELP,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument("--version", action="version", version="%(prog)s " + VERSION)
parser.add_argument("--base-url", help="default: HONEYLOG_BASE_URL, else the address of the last login, else " + DEFAULT_BASE_URL)
parser.add_argument("--token", help="API token instead of the login; prefer HONEYLOG_TOKEN, since flags show up in the process list")
parser.add_argument("--format", choices=["table", "json", "jsonl", "csv"], help="default: table on a terminal, json otherwise")
parser.add_argument("--timeout", type=float, default=60.0, help="seconds per request (default: 60)")
parser.add_argument("--retries", type=int, default=3, help="retries on rate limits and network errors (default: 3)")
parser.add_argument("--no-retry", action="store_true", help="fail on the first rate limit or network error")
parser.add_argument("--allow-http", action="store_true", help="allow a plain http base URL outside localhost")
parser.add_argument("-v", "--verbose", action="store_true", help="print each request (never the token) to stderr")
commands = parser.add_subparsers(dest="command", metavar="command")
commands.required = True
login = commands.add_parser("login", help="sign in with the browser, once: the token is saved for the next terminals")
login.add_argument("--no-browser", action="store_true", help="print the sign-in address instead of opening a browser, and take the answer pasted back (for a remote machine)")
commands.add_parser("logout", help="revoke the saved token and remove it from this computer")
sites = commands.add_parser("sites", help="the site groups the token can read, with their wireable")
sites.add_argument("--search", help="part of a domain or network name, or a site uuid")
sites.add_argument("--type", choices=["sites", "networks"])
sites.add_argument("--team-id", type=int, help="only the site groups of this team")
_add_paging(sites, max_rows_default=10000)
fields = commands.add_parser("fields", help="the metric, dimension and column names a site group accepts")
fields.add_argument("wireable")
fields.add_argument("--source", choices=["stats", "raw_logs"], help="raw_logs for the raw-logs columns (default: stats)")
timeseries = commands.add_parser("timeseries", help="metrics over time (GET /stats/views)")
timeseries.add_argument("wireable")
_add_window(timeseries, STATS_RANGES, STATS_RANGE_HELP, compare=True)
timeseries.add_argument("--granularity", choices=["minute", "hour", "day", "week", "month", "year"])
timeseries.add_argument("--metrics", help="comma separated, from the fields command")
_add_filters(timeseries, having=True)
breakdown = commands.add_parser("breakdown", help="metrics per dimension value (GET /stats/pages)")
breakdown.add_argument("wireable")
_add_window(breakdown, STATS_RANGES, STATS_RANGE_HELP, compare=True)
breakdown.add_argument("--dimension", help="default: pathname")
breakdown.add_argument("--dimensions", help="further dimensions, comma separated")
breakdown.add_argument("--metrics", help="comma separated, from the fields command")
breakdown.add_argument("--sort", help="a requested metric or the dimension")
breakdown.add_argument("--sort-order", choices=["asc", "desc"])
breakdown.add_argument("--with-totals", action="store_true", help="also the metrics over every matching row, not only this page (meta.totals in json, under the table)")
_add_filters(breakdown, having=True)
_add_paging(breakdown, max_rows_default=10000)
raw_logs = commands.add_parser("raw-logs", help="the individual requests (GET /raw-logs)")
raw_logs.add_argument("wireable")
_add_window(raw_logs, RAW_LOGS_RANGES, RAW_LOGS_RANGE_HELP, compare=False)
raw_logs.add_argument("--last", help="the window up to now, such as 30m, 6h or 2d (instead of --start/--end)")
raw_logs.add_argument("--columns", help="comma separated, from: fields WIREABLE --source raw_logs")
raw_logs.add_argument("--order", "--order-direction", dest="order", choices=["asc", "desc"], help="default: desc (newest first)")
_add_filters(raw_logs, having=False)
_add_paging(raw_logs, max_rows_default=10000)
raw_logs_timeseries = commands.add_parser("raw-logs-timeseries", aliases=["hits"], help="requests per minute, hour or day (GET /raw-logs/timeseries)")
raw_logs_timeseries.add_argument("wireable")
_add_window(raw_logs_timeseries, RAW_LOGS_RANGES, RAW_LOGS_RANGE_HELP, compare=False)
raw_logs_timeseries.add_argument("--last", help="the window up to now, such as 30m, 6h or 2d (instead of --start/--end)")
raw_logs_timeseries.add_argument(
"--granularity",
choices=RAW_LOGS_GRANULARITIES,
help="bucket size: 1m on windows up to 4 hours, 5m up to 24 hours, 15m up to 72 hours, 1h and 1d on any window (default: the finest available)",
)
_add_filters(raw_logs_timeseries, having=False)
return parser
def _add_window(parser: argparse.ArgumentParser, ranges: Sequence[str], range_help: str, compare: bool) -> None:
parser.add_argument("--range", choices=ranges, metavar="PERIOD", help="a named period instead of --start/--end, in the site timezone ({}): {}".format(range_help, ", ".join(ranges)))
parser.add_argument("--start", help="epoch seconds or a date such as 2026-09-01 or 2026-09-01T10:00:00Z, in the site timezone")
parser.add_argument("--end", help="same formats as --start (default: now)")
if compare:
parser.add_argument("--compare-range", choices=COMPARE_RANGES, metavar="PERIOD", help="a window to compare against, relative to the main one: " + ", ".join(COMPARE_RANGES))
parser.add_argument("--compare-start", help="start of a window to compare against, for dates --compare-range cannot express")
parser.add_argument("--compare-end", help="end of the comparison window")
def _add_filters(parser: argparse.ArgumentParser, having: bool) -> None:
parser.add_argument("--where", "--where-expression", dest="where", help='filter as JSON, @file or - for stdin: [[["pathname","=","/pricing"]]]')
if having:
parser.add_argument("--having", "--having-expression", dest="having", help='filter on metric values, same forms: [[["views",">",100]]]')
def _add_paging(parser: argparse.ArgumentParser, max_rows_default: int) -> None:
parser.add_argument("--limit", type=int, help="rows per page")
parser.add_argument("--offset", type=int, help="rows to skip")
parser.add_argument("--all", action="store_true", help="follow the pages up to --max-rows")
parser.add_argument("--max-rows", type=int, default=max_rows_default, help="stop --all after this many rows (default: {})".format(max_rows_default))
def describe_commands() -> Dict[str, Dict[str, Any]]:
"""Every command with its API route, its options and their allowed values, read from the parser."""
parser = build_parser()
subparsers = next(action for action in parser._actions if isinstance(action, argparse._SubParsersAction))
commands: Dict[str, Dict[str, Any]] = {}
for name, command in subparsers.choices.items():
# Aliases such as hits point at a parser already listed under its own
# name, and login and logout call no API route.
if name not in COMMAND_ENDPOINTS:
continue
options: List[str] = []
choices: Dict[str, List[str]] = {}
positionals: List[str] = []
for action in command._actions:
if not action.option_strings:
positionals.append(action.dest)
continue
options.extend(action.option_strings)
for option in action.option_strings if action.choices else []:
choices[option] = list(action.choices)
commands[name] = {
"endpoint": COMMAND_ENDPOINTS[name],
"wireable": "wireable" in positionals,
"options": options,
"choices": choices,
}
return commands
def check_base_url(base_url: str, allow_http: bool) -> str:
parts = urllib.parse.urlsplit(base_url)
if parts.scheme not in ("http", "https") or not parts.hostname:
raise UsageError("--base-url must be an http(s) URL, such as " + DEFAULT_BASE_URL)
if parts.scheme == "http" and parts.hostname not in LOOPBACK_HOSTS and not allow_http:
raise UsageError("refusing to send the token over plain http to {}: use https, or --allow-http".format(parts.hostname))
return base_url
def main(
argv: Optional[Sequence[str]] = None,
stdout: Optional[TextIO] = None,
stderr: Optional[TextIO] = None,
stdin: Optional[TextIO] = None,
env: Optional[Dict[str, str]] = None,
sleep: Callable[[float], None] = time.sleep,
open_browser: Callable[[str], bool] = webbrowser.open,
) -> int:
stdout = stdout or sys.stdout
stderr = stderr or sys.stderr
stdin = stdin or sys.stdin
env = dict(os.environ) if env is None else env
parser = build_parser()
try:
args = parser.parse_args(argv)
except SystemExit as exit_:
return int(exit_.code or 0)
if args.format is None:
args.format = "table" if getattr(stdout, "isatty", lambda: False)() else "json"
def warn(message: str) -> None:
stderr.write("honeylog: " + message + "\n")
token_source = ""
try:
store = CredentialStore(os.path.join(config_dir(env), CREDENTIALS_FILE), warn)
base_url = check_base_url(
args.base_url or env.get("HONEYLOG_BASE_URL") or store.default_base_url() or DEFAULT_BASE_URL,
args.allow_http,
)
if args.command == "login":
command_login(args, base_url, store, env, stdin, stderr, warn, open_browser)
return EXIT_OK
if args.command == "logout":
command_logout(args, base_url, store, env, warn)
return EXIT_OK
token, token_source = resolve_token(args, env, store, base_url)
if not token:
raise UsageError("not logged in to {}: run honeylog login, or set HONEYLOG_TOKEN".format(base_url))
if token_source == "login":
check_login_expiry(store, base_url, stderr, warn)
if args.verbose:
stderr.write("token from {}\n".format("the login saved in " + store.path if token_source == "login" else token_source))
client = Client(
base_url,
token,
timeout=args.timeout,
retries=0 if args.no_retry else args.retries,
verbose=args.verbose,
stderr=stderr,
sleep=sleep,
)
if args.command == "sites":
command_sites(client, args, stdout, warn)
elif args.command == "fields":
command_fields(client, args, stdout, warn)
elif args.command == "timeseries":
command_timeseries(client, args, stdout, warn, stdin)
elif args.command == "breakdown":
command_breakdown(client, args, stdout, warn, stdin)
elif args.command == "raw-logs":
command_raw_logs(client, args, stdout, warn, stdin)
elif args.command in ("raw-logs-timeseries", "hits"):
command_raw_logs_timeseries(client, args, stdout, warn, stdin)
# A short output is still in the buffer: flush it here, where a closed
# pipe lands in the handler below rather than in Python's exit.
stdout.flush()
return EXIT_OK
except UsageError as error:
warn(str(error))
return EXIT_USAGE
except LoginError as error:
warn(str(error))
return EXIT_AUTH
except ApiError as error:
warn("HTTP {}: {}".format(error.status, error.message))
for field, messages in error.errors.items():
for message in messages if isinstance(messages, list) else [messages]:
stderr.write(" {}: {}\n".format(field, message))
if error.status == 401 and token_source == "login":
warn("the saved login no longer works, maybe its token was deleted: run honeylog login again")
return error.exit_code
except NetworkError as error:
warn(str(error))
return EXIT_NETWORK
except KeyboardInterrupt:
stderr.write("\n")
return EXIT_INTERRUPTED
except BrokenPipeError:
# Piped into head or similar, which closed the pipe early: stop quietly,
# and keep Python from complaining again when it flushes stdout at exit.
try:
os.dup2(os.open(os.devnull, os.O_WRONLY), sys.stdout.fileno())
except (OSError, ValueError):
pass
return EXIT_OK
if __name__ == "__main__":
sys.exit(main())