j3d1/toolshed
2
1
Fork 0
Code Issues 9 Pull requests 1 Projects Releases Packages Wiki Activity

stash

Browse source
This commit is contained in:
j3d1 2026-08-27 03:55:02 +02:00
parent fdb83f65d4
commit 8cba1f52e8
1 changed files with 90 additions and 36 deletions
Download patch file Download diff file Expand all files Collapse all files

126
cli-client/toolshed-client.py
View file

@ -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__':