From 8cba1f52e8826d36cc779f856de32d2c780143a2 Mon Sep 17 00:00:00 2001 From: jedi Date: Thu, 27 Aug 2026 03:55:02 +0200 Subject: [PATCH] stash --- cli-client/toolshed-client.py | 126 ++++++++++++++++++++++++---------- 1 file changed, 90 insertions(+), 36 deletions(-) diff --git a/cli-client/toolshed-client.py b/cli-client/toolshed-client.py index 542fe2f..cdbeffe 100755 --- a/cli-client/toolshed-client.py +++ b/cli-client/toolshed-client.py @@ -14,70 +14,107 @@ COMMANDS = { } +class ApiError(Exception): + """A problem talking to the backend - network/TLS failure, or a response that isn't the JSON + we expected. Deliberately distinct from ValueError (bad input) so main() can report both with + a plain message instead of a traceback, without conflating "you gave me something invalid" + with "the server didn't behave".""" + + class ToolshedApi: user = None host = None signing_key = None - def __init__(self, user, host, key): + def __init__(self, user, host, key, ca_cert=None): if host is None: - raise ValueError("TOOLSHED_HOST environment variable not set") + raise ValueError("No host configured - set TOOLSHED_HOST or pass --host (e.g. a.localhost:8000)") if user is None: - raise ValueError("TOOLSHED_USER environment variable not set") + raise ValueError("No user configured - set TOOLSHED_USER or pass --user (e.g. you@a.localhost)") if key is None: - raise ValueError("TOOLSHED_KEY environment variable not set") + raise ValueError("No signing key configured - set TOOLSHED_KEY or pass --key") if len(key) != 64: - raise ValueError("TOOLSHED_KEY must be 64 hex characters") + raise ValueError("TOOLSHED_KEY must be 64 hex characters, got {} characters".format(len(key))) - signing_key = SigningKey(bytes.fromhex(key)) + try: + signing_key = SigningKey(bytes.fromhex(key)) + except ValueError: + raise ValueError("TOOLSHED_KEY must be a hex-encoded Ed25519 private key") + + if ca_cert is not None and not os.path.isfile(ca_cert): + raise ValueError("CA cert file not found: {}".format(ca_cert)) self.user = user self.host = host self.signing_key = signing_key + self.verify = ca_cert if ca_cert is not None else True self._spec = None + def _url(self, target): + return "https://" + self.host + target + + def _ssl_error(self, error): + hint = "" if self.verify is not True else \ + " - if this is a dev server with a self-signed cert, pass --ca-cert/TOOLSHED_CA_CERT" + return ApiError("TLS error talking to {}: {}{}".format(self.host, error, hint)) + + def _send(self, method, target, json_body=None): + url = self._url(target) + signed_body = dumps(json_body).encode('utf-8') if json_body is not None else b'' + signature = self.signing_key.sign(url.encode('utf-8') + signed_body).signature.hex() + headers = {"Authorization": "Signature " + self.user + ":" + signature} + try: + return requests.request(method, url, headers=headers, json=json_body, verify=self.verify) + except requests.exceptions.SSLError as error: + raise self._ssl_error(error) + except requests.exceptions.ConnectionError as error: + raise ApiError("Could not reach {} - is the host/port correct and reachable? ({})".format( + self.host, error)) + except requests.exceptions.Timeout: + raise ApiError("Request to {} timed out".format(self.host)) + + @staticmethod + def _parse_json(response): + try: + return response.json() + except ValueError: + raise ApiError("Expected a JSON response from {} but got {} {}: {}".format( + response.url, response.status_code, response.reason, response.text[:300])) + def get_spec(self): if self._spec is None: - response = requests.get("http://" + self.host + "/docs/?format=openapi") - response.raise_for_status() - self._spec = response.json() + try: + response = requests.get(self._url("/docs/?format=openapi"), verify=self.verify) + except requests.exceptions.SSLError as error: + raise self._ssl_error(error) + except requests.exceptions.ConnectionError as error: + raise ApiError("Could not reach {} - is the host/port correct and reachable? ({})".format( + self.host, error)) + self._spec = self._parse_json(response) return self._spec def get(self, target): - url = "http://" + self.host + target - signed = self.signing_key.sign(url.encode('utf-8')) - signature = signed.signature.hex() - response = requests.get(url, headers={"Authorization": "Signature " + self.user + ":" + signature}) - return response.json() + return self._parse_json(self._send('GET', target)) def post(self, target, data): - url = "http://" + self.host + target - json = dumps(data) - signed = self.signing_key.sign(url.encode('utf-8') + json.encode('utf-8')) - signature = signed.signature.hex() - response = requests.post(url, headers={"Authorization": "Signature " + self.user + ":" + signature}, json=data) - return response.json() + return self._parse_json(self._send('POST', target, json_body=data)) def delete(self, target): - url = "http://" + self.host + target - signed = self.signing_key.sign(url.encode('utf-8')) - signature = signed.signature.hex() - response = requests.delete(url, headers={"Authorization": "Signature " + self.user + ":" + signature}) + response = self._send('DELETE', target) if not response.content: return {"deleted": response.ok} - return response.json() + return self._parse_json(response) def get_raw(self, target): """Like get(), but returns the raw response body instead of parsing it as JSON - for endpoints like /api/export/ that hand back a zip file, not a JSON document.""" - url = "http://" + self.host + target - signed = self.signing_key.sign(url.encode('utf-8')) - signature = signed.signature.hex() - response = requests.get(url, headers={"Authorization": "Signature " + self.user + ":" + signature}) - response.raise_for_status() + response = self._send('GET', target) + if not response.ok: + raise ApiError("{} {} from {}: {}".format( + response.status_code, response.reason, response.url, response.text[:300])) return response.content @@ -214,9 +251,15 @@ def format_table(rows): for key in row.keys(): if key not in columns: columns.append(key) - lines = ['| ' + ' | '.join(columns) + ' |', '| ' + ' | '.join('---' for _ in columns) + ' |'] - for row in rows: - lines.append('| ' + ' | '.join(stringify_cell(row.get(column)) for column in columns) + ' |') + cells = [[stringify_cell(row.get(column)) for column in columns] for row in rows] + widths = [max([len(columns[i])] + [len(cell[i]) for cell in cells] + [3]) for i in range(len(columns))] + + def format_row(values): + return '| ' + ' | '.join(value.ljust(widths[i]) for i, value in enumerate(values)) + ' |' + + lines = [format_row(columns), '|-' + '-|-'.join('-' * width for width in widths) + '-|'] + for cell in cells: + lines.append(format_row(cell)) return '\n'.join(lines) @@ -235,11 +278,15 @@ def main(): host = os.environ.get('TOOLSHED_HOST') user = os.environ.get('TOOLSHED_USER') key = os.environ.get('TOOLSHED_KEY') + ca_cert = os.environ.get('TOOLSHED_CA_CERT') parser = argparse.ArgumentParser(description='Toolshed API client') parser.add_argument('--host', help='Toolshed host') parser.add_argument('--user', help='Toolshed user') parser.add_argument('--key', help='Toolshed key') + parser.add_argument('--ca-cert', + help='CA cert file to verify the server against, for a dev/self-signed host ' + '(e.g. frontend/.local/RootCA.crt) - normal system CAs are used otherwise') parser.add_argument('--json', action='store_true', help='Read input as JSON from stdin and print output as JSON, instead of prompting/printing') parser.add_argument('cmd', help='Command') @@ -255,11 +302,18 @@ def main(): if args.key is not None: key = args.key - api = ToolshedApi(user, host, key) + if args.ca_cert is not None: + ca_cert = args.ca_cert - json_input = read_json_input() if args.json else None + try: + api = ToolshedApi(user, host, key, ca_cert) + json_input = read_json_input() if args.json else None + result = run_command(api, args.cmd, args.args, json_input) + except (ValueError, ApiError, OSError) as error: + print("Error: {}".format(error), file=sys.stderr) + sys.exit(1) - print_result(run_command(api, args.cmd, args.args, json_input), args.json) + print_result(result, args.json) if __name__ == '__main__':