Snapshot: alpha-2026-9

cli-client/toolshed-client.py

@@ -0,0 +1,320 @@
+#!/usr/bin/env python3
+import argparse
+import base64
+import os
+import sys
+import requests
+from nacl.signing import SigningKey
+from json import dumps, loads
+
+COMMANDS = {
+    'getinventory': {'path': '/api/v1/inventory_items/{handle}/', 'method': 'get'},
+    'additem': {'path': '/api/v1/inventory_items/{handle}/', 'method': 'post'},
+    'delitem': {'path': '/api/v1/inventory_items/{handle}/{internal_id}/', 'method': 'delete'},
+}
+
+
+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, ca_cert=None):
+        if host is None:
+            raise ValueError("No host configured - set TOOLSHED_HOST or pass --host (e.g. a.localhost:8000)")
+
+        if user is None:
+            raise ValueError("No user configured - set TOOLSHED_USER or pass --user (e.g. you@a.localhost)")
+
+        if key is None:
+            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, got {} characters".format(len(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:
+            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):
+        return self._parse_json(self._send('GET', target))
+
+    def post(self, target, data):
+        return self._parse_json(self._send('POST', target, json_body=data))
+
+    def delete(self, target):
+        response = self._send('DELETE', target)
+        if not response.content:
+            return {"deleted": response.ok}
+        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/v1/export/ that hand back a zip file, not a JSON document."""
+        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
+
+
+def resolve_schema(spec, schema):
+    if '$ref' in schema:
+        return spec['definitions'][schema['$ref'].split('/')[-1]]
+    return schema
+
+
+def operation_parameters(spec, path, method):
+    """Path- and body-parameters for an operation, per the endpoint's own OpenAPI spec entry."""
+    path_item = spec['paths'][path]
+    params = path_item.get('parameters', []) + path_item[method].get('parameters', [])
+    path_params = [p for p in params if p.get('in') == 'path']
+    body_param = next((p for p in params if p.get('in') == 'body'), None)
+    return path_params, body_param
+
+
+def parse_kv_args(cmd_args):
+    """['name=Drill', 'owned_quantity=2'] -> {'name': 'Drill', 'owned_quantity': '2'}"""
+    pairs = {}
+    for arg in cmd_args:
+        if '=' in arg:
+            key, value = arg.split('=', 1)
+            pairs[key] = value
+    return pairs
+
+
+def resolve_path_params(api, path_params, cmd_args, json_input):
+    values = {}
+    positional = [arg for arg in cmd_args if '=' not in arg]
+    kv = parse_kv_args(cmd_args)
+    for param in path_params:
+        name = param['name']
+        if name == 'handle':
+            values[name] = api.user
+            continue
+        if json_input is not None:
+            value = json_input.get(name, json_input.get('id'))
+            if value is None:
+                raise ValueError("Missing required path parameter '{}' in --json stdin input".format(name))
+        elif name in kv:
+            value = kv[name]
+        elif 'id' in kv:
+            value = kv['id']
+        elif positional:
+            value = positional.pop(0)
+        else:
+            value = input("{}: ".format(name))
+        values[name] = value
+    return values
+
+
+def resolve_body(spec, body_param, cmd_args, json_input):
+    if body_param is None:
+        return None
+    if json_input is not None:
+        return json_input
+    kv = parse_kv_args(cmd_args)
+    if kv:
+        return kv
+    schema = resolve_schema(spec, body_param['schema'])
+    writable_fields = [name for name, prop in schema.get('properties', {}).items() if not prop.get('readOnly')]
+    body = {}
+    for field in writable_fields:
+        value = input("{}: ".format(field))
+        if value != '':
+            body[field] = value
+    return body
+
+
+def build_request(api, spec, command, cmd_args, json_input):
+    path_params, body_param = operation_parameters(spec, command['path'], command['method'])
+    values = resolve_path_params(api, path_params, cmd_args, json_input)
+    url = command['path']
+    for name, value in values.items():
+        url = url.replace('{' + name + '}', str(value))
+    body = resolve_body(spec, body_param, cmd_args, json_input)
+    return url, body
+
+
+def run_command(api, cmd, cmd_args, json_input):
+    if cmd == 'export':
+        path = cmd_args[0] if cmd_args else 'toolshed-export.zip'
+        data = api.get_raw("/api/v1/export/")
+        with open(path, 'wb') as f:
+            f.write(data)
+        return {'exported_to': path, 'bytes': len(data)}
+    elif cmd == 'import':
+        if not cmd_args:
+            raise ValueError("import requires the path to a previously exported zip file")
+        path = cmd_args[0]
+        with open(path, 'rb') as f:
+            data = f.read()
+        return api.post("/api/v1/import/", {"zip": base64.b64encode(data).decode('ascii')})
+
+    command = COMMANDS.get(cmd)
+    if command is None:
+        raise ValueError("Unknown command: " + cmd)
+    spec = api.get_spec()
+    url, body = build_request(api, spec, command, cmd_args, json_input)
+    if command['method'] == 'get':
+        return api.get(url)
+    elif command['method'] == 'post':
+        return api.post(url, body or {})
+    elif command['method'] == 'delete':
+        return api.delete(url)
+    else:
+        raise ValueError("Unsupported method: " + command['method'])
+
+
+def read_json_input():
+    if sys.stdin.isatty():
+        return {}
+    raw = sys.stdin.read()
+    return loads(raw) if raw.strip() else {}
+
+
+def stringify_cell(value):
+    if value is None:
+        return ''
+    if isinstance(value, (list, tuple)):
+        return ', '.join(stringify_cell(v) for v in value)
+    if isinstance(value, dict):
+        return dumps(value)
+    return str(value)
+
+
+def format_table(rows):
+    if not rows:
+        return '(empty)'
+    columns = []
+    for row in rows:
+        for key in row.keys():
+            if key not in columns:
+                columns.append(key)
+    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)
+
+
+def print_result(result, as_json):
+    if as_json:
+        print(dumps(result, indent=2))
+    elif isinstance(result, list) and all(isinstance(item, dict) for item in result):
+        print(format_table(result))
+    elif isinstance(result, dict):
+        print(format_table([result]))
+    else:
+        print(result)
+
+
+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')
+    parser.add_argument('args', nargs='*', help="Command arguments, as key=value pairs (e.g. name=Drill)")
+    args = parser.parse_args()
+
+    if args.host is not None:
+        host = args.host
+
+    if args.user is not None:
+        user = args.user
+
+    if args.key is not None:
+        key = args.key
+
+    if args.ca_cert is not None:
+        ca_cert = args.ca_cert
+
+    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(result, args.json)
+
+
+if __name__ == '__main__':
+    main()