#!/usr/bin/env python3 import argparse import json import sys import urllib.parse from collections.abc import Sequence from typing import Union import argcomplete import requests from argcomplete.completers import subprocess from requests.models import HTTPBasicAuth base_url = "https://knotapitest.sunet.se" user = "kano" password = subprocess.getoutput("pass show knotapitest.sunet.se/{}".format(user)) # Authenticate basic = HTTPBasicAuth(user, password) response = requests.get(base_url + "/user/login", auth=basic) token = response.json()["token"] headers = {"Authorization": "Bearer {}".format(token)} # Parse out information for each command line option parser = argparse.ArgumentParser() parser.add_argument("--json", action=argparse.BooleanOptionalAction) subparsers = parser.add_subparsers(dest="command") add = subparsers.add_parser("add") add.add_argument( "-d", "--data", nargs="*", help="Specify any number of key - value pairs: name=dns1.example.com.", ) add.add_argument("--name") add.add_argument("--rtype") add.add_argument("--zone", required=True) delete = subparsers.add_parser("delete") delete.add_argument( "--data", nargs="*", help="Specify any number of key - value pairs: name=dns1.example.com.", ) delete.add_argument("--name") delete.add_argument("--rtype") delete.add_argument("--zone", required=True) listcmd = subparsers.add_parser("list") listcmd.add_argument( "--data", nargs="*", help="Specify any number of key - value pairs: name=dns1.example.com.", ) listcmd.add_argument("--name") listcmd.add_argument("--rtype") listcmd.add_argument("--zone") update = subparsers.add_parser("update") update.add_argument( "--data", nargs="*", help="Specify any number of key - value pairs: name=dns1.example.com.", required=True, ) update.add_argument("--name", required=True) update.add_argument("--rtype", required=True) update.add_argument("--zone", required=True) argcomplete.autocomplete(parser) args = parser.parse_args() def error(description: str, error: str) -> Sequence[dict]: response = [] reply = {} # https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/406 reply["Code"] = 406 reply["Description"] = description reply["Error"] = error response.append(reply) return response def output(response: Sequence[dict], jsonout: bool = False): try: if jsonout: print(json.dumps(response)) else: for entry in response: print(nested_out(entry)) except BrokenPipeError: pass def nested_out(input, tabs="") -> str: string = "" if isinstance(input, str): string += "{}\n".format(input) elif isinstance(input, dict): for key, value in input.items(): string += "{}{}: {}".format(tabs, key, nested_out(value, tabs + " ")) elif isinstance(input, list): for entry in input: string += "{}\n{}".format(tabs, nested_out(entry, tabs + " ")) return string # Define the parser for each command def run_add(url: str, jsonout: bool): response = requests.put(url, headers=headers) output(response.json(), jsonout) def run_delete(url: str, jsonout: bool): response = requests.delete(url, headers=headers) output(response.json(), jsonout) def run_list(url: str, jsonout: bool): response = requests.get(url, headers=headers) output(response.json(), jsonout) def run_update(url: str, jsonout: bool): response = requests.patch(url, headers=headers) output(response.json(), jsonout) # Set up the url def setup_url( data: Union[None, Sequence[str]], name: Union[None, str], rtype: Union[None, str], zone: Union[None, str], ) -> str: url = base_url + "/zones" if zone: url += "/{}".format(zone) if name and zone: url += "/records/{}".format(name) if zone and name and rtype: url += "/{}".format(rtype) if data and zone and name and rtype: url += "?" for arg in data: if not url.endswith("?"): url += "&" url += urllib.parse.quote_plus(arg) if rtype and (not name or not zone): output(error("rtype only makes sense with name and zone", "Missing parameter")) sys.exit(1) if name and not zone: output(error("name only makes sense with a zone", "Missing parameter")) sys.exit(1) return url # Route based on command url = setup_url(args.data, args.name, args.rtype, args.zone) if args.command == "add": run_add(url, args.json) elif args.command == "delete": run_delete(url, args.json) elif args.command == "list": run_list(url, args.json) elif args.command == "update": run_update(url, args.json)