parent
33a65181a5
commit
d1aec6bbb0
GPG Key ID:
0DA0A7A5708FE257
@ -0,0 +1,61 @@
|
||||
#!/usr/bin/env python3
|
||||
import os
|
||||
from os import mkdir
|
||||
from os.path import isdir, isfile, join
|
||||
from typing import Union
|
||||
|
||||
import yaml
|
||||
|
||||
|
||||
class Config:
|
||||
|
||||
def __init__(self):
|
||||
# Make sure we have config
|
||||
self.config_basepath = join(os.environ["HOME"], ".knot")
|
||||
self.config_filename = join(self.config_basepath, "config")
|
||||
if not isdir(self.config_basepath):
|
||||
mkdir(self.config_basepath)
|
||||
|
||||
def get_config(self) -> Union[None, dict]:
|
||||
if not isfile(self.config_filename):
|
||||
return None
|
||||
with open(self.config_filename, "r") as fh:
|
||||
return yaml.safe_load(fh.read())
|
||||
|
||||
def get_config_data(self) -> dict:
|
||||
config_data = self.get_config()
|
||||
config_data.pop("password", None)
|
||||
return config_data
|
||||
|
||||
def get_current(self) -> str:
|
||||
if os.path.islink(self.config_filename):
|
||||
actual_path = os.readlink(self.config_filename)
|
||||
return actual_path.split("-")[-1]
|
||||
else:
|
||||
return "none"
|
||||
|
||||
def set_context(self, context) -> bool:
|
||||
symlink = f"{self.config_filename}-{context}"
|
||||
found = os.path.isfile(symlink)
|
||||
if os.path.islink(self.config_filename):
|
||||
os.remove(self.config_filename)
|
||||
elif os.path.isfile(self.config_filename):
|
||||
os.rename(self.config_filename, symlink)
|
||||
os.symlink(symlink, self.config_filename)
|
||||
self.config_filename = symlink
|
||||
return found
|
||||
|
||||
def set_config(
|
||||
self,
|
||||
baseurl: str,
|
||||
username: str,
|
||||
password: str,
|
||||
):
|
||||
config = {
|
||||
"baseurl": baseurl,
|
||||
"username": username,
|
||||
"password": password
|
||||
}
|
||||
|
||||
with open(self.config_filename, "w") as fh:
|
||||
fh.write(yaml.dump(config))
|
||||
@ -0,0 +1,17 @@
|
||||
import openstack
|
||||
import openstack.config.loader
|
||||
|
||||
|
||||
def get_openstack_addresses(cloud: str, name: str):
|
||||
conn = openstack.connect(cloud=cloud)
|
||||
|
||||
# List the servers
|
||||
server = conn.compute.find_server(name)
|
||||
if server is None:
|
||||
print("Server not found")
|
||||
exit(1)
|
||||
openstack_addresses = []
|
||||
for network in server.addresses:
|
||||
for address in server.addresses[network]:
|
||||
openstack_addresses.append(address)
|
||||
return openstack_addresses
|
||||
@ -0,0 +1,123 @@
|
||||
import json
|
||||
import sys
|
||||
import urllib.parse
|
||||
from typing import Union
|
||||
from urllib.parse import urlparse
|
||||
|
||||
|
||||
def error(description: str, error: str):
|
||||
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 nested_out(input, tabs="") -> str:
|
||||
string = ""
|
||||
if isinstance(input, str) or isinstance(input, int):
|
||||
string += f"{input}\n"
|
||||
elif isinstance(input, dict):
|
||||
for key, value in input.items():
|
||||
string += f"{tabs}{key}: {nested_out(value, tabs + " ")}"
|
||||
elif isinstance(input, list):
|
||||
for entry in input:
|
||||
string += f"{tabs}\n{nested_out(entry, tabs + ' ')}"
|
||||
return string
|
||||
|
||||
|
||||
def output(response: list[dict], jsonout: bool = False):
|
||||
try:
|
||||
if jsonout:
|
||||
print(json.dumps(response))
|
||||
else:
|
||||
print(nested_out(response))
|
||||
except BrokenPipeError:
|
||||
pass
|
||||
|
||||
|
||||
def setup_url(
|
||||
baseurl: str,
|
||||
arguments: Union[None, list[str]],
|
||||
data: Union[None, str],
|
||||
name: Union[None, str],
|
||||
rtype: Union[None, str],
|
||||
ttl: Union[None, str],
|
||||
zone: Union[None, str],
|
||||
) -> str:
|
||||
url = baseurl + "/zones"
|
||||
if zone:
|
||||
if not zone.endswith("."):
|
||||
zone += "."
|
||||
url += "/{}".format(zone)
|
||||
if name and zone:
|
||||
if name.endswith(zone.rstrip(".")):
|
||||
name += "."
|
||||
url += "/records/{}".format(name)
|
||||
if zone and name and rtype:
|
||||
url += "/{}".format(rtype)
|
||||
if data and zone and name and rtype:
|
||||
url += "/{}".format(data)
|
||||
if ttl and data and zone and name and rtype:
|
||||
url += "/{}".format(ttl)
|
||||
if data and zone and name and rtype and arguments:
|
||||
url += "?"
|
||||
for arg in arguments:
|
||||
if not url.endswith("?"):
|
||||
url += "&"
|
||||
key, value = arg.split("=")
|
||||
url += key + "=" + urllib.parse.quote_plus(value)
|
||||
|
||||
if ttl and (not rtype or not name or not zone):
|
||||
output(
|
||||
error(
|
||||
"ttl only makes sense with rtype, name and zone",
|
||||
"Missing parameter",
|
||||
))
|
||||
sys.exit(1)
|
||||
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
|
||||
|
||||
|
||||
def split_url(url: str) -> dict:
|
||||
parsed = urlparse(url, allow_fragments=False)
|
||||
path = parsed.path
|
||||
query = parsed.query
|
||||
arguments: Union[None, list[str]] = query.split("&")
|
||||
path_arr = path.split("/")
|
||||
data: Union[None, str] = None
|
||||
name: Union[None, str] = None
|
||||
rtype: Union[None, str] = None
|
||||
ttl: Union[None, str] = None
|
||||
zone: Union[None, str] = None
|
||||
if len(path_arr) > 2:
|
||||
zone = path_arr[2]
|
||||
if len(path_arr) > 4:
|
||||
name = path_arr[4]
|
||||
if len(path_arr) > 5:
|
||||
rtype = path_arr[5]
|
||||
if len(path_arr) > 6:
|
||||
data = path_arr[6]
|
||||
if len(path_arr) > 7:
|
||||
ttl = path_arr[7]
|
||||
|
||||
return {
|
||||
"arguments": arguments,
|
||||
"data": data,
|
||||
"name": name,
|
||||
"rtype": rtype,
|
||||
"ttl": ttl,
|
||||
"zone": zone,
|
||||
}
|
||||
Loading…
Reference in new issue