misc/inwx
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: misc:281e38217523659d7bec76f44f79fdddb5de2e78
Branches Tags
misc:main
..
compare: misc:513657c1fec74836f8b796e628c647a91ebb312b
Branches Tags
misc:main

No commits in common. "281e38217523659d7bec76f44f79fdddb5de2e78" and "513657c1fec74836f8b796e628c647a91ebb312b" have entirely different histories.
281e382175 ... 513657c1fe

6 changed files with 152 additions and 439 deletions
Show stats Expand all files Collapse all files

1
.gitignore vendored
View file

@ -1,2 +1 @@
build/ build/
/.geany

148
source/conf.py
View file

@ -1,121 +1,38 @@
_conf_data = None _conf_data = {
"url": {
"test": {
def conf_schema( "scheme": "https",
): "host": "api.ote.domrobot.com",
return { "port": 443,
"type": "object", "path": "jsonrpc/"
"properties": {
"url": {
"type": "object",
"properties": {
},
"additionalProperties": {
"type": "object",
"properties": {
"scheme": {
"type": "string",
},
"host": {
"type": "string",
},
"port": {
"type": "number",
},
"path": {
"type": "string",
},
},
"additionalProperties": False,
"required": [
"host",
]
},
"required": [
]
},
"environment": {
"type": "string",
},
"account": {
"type": "object",
"properties": {
"username": {
"type": "string",
},
"password": {
"type": "string",
},
},
"additionalProperties": False,
"required": [
]
}
}, },
"additionalProperties": False, "production": {
"required": [ "scheme": "https",
], "host": "api.domrobot.com",
"port": 443,
"path": "jsonrpc/"
}
},
"environment": "production",
"account": {
"username": None,
"password": None
} }
}
def conf_load( def conf_load(
path : str path : str
): ):
global _conf_data global _conf_data
conf_data_raw = ( if (not _os.path.exists(path)):
_json.loads(file_read(path)) pass
if _os.path.exists(path) else else:
{} handle = open(path, "r")
) content = handle.read()
for pair in conf_data_raw.get("url", {}).items(): handle.close()
if ("host" in pair[1]): data = _json.loads(content)
pass _conf_data = merge(_conf_data, data)
else:
raise ValueError("flawed conf: missing mandatory value 'host' for url entry '%s'" % pair[0])
_conf_data = {
"url": convey(
(
{
"test": {
"scheme": "https",
"host": "api.ote.domrobot.com",
"port": 443,
"path": "jsonrpc/"
},
"production": {
"scheme": "https",
"host": "api.domrobot.com",
"port": 443,
"path": "jsonrpc/"
}
}
|
conf_data_raw.get("url", {})
),
[
lambda x: x.items(),
lambda pairs: map(
lambda pair: (
pair[0],
{
"scheme": pair[1].get("scheme", "https"),
"host": pair[1]["host"],
"port": pair[1].get("port", 443),
"path": pair[1].get("path", "jsonrpc/"),
}
),
pairs
),
dict,
]
),
"environment": conf_data_raw.get("environment", "production"),
"account": {
"username": conf_data_raw.get("account", {}).get("username", None),
"password": conf_data_raw.get("account", {}).get("password", None),
}
}
# print(_json.dumps(_conf_data, indent = "\t"))
def conf_get( def conf_get(
@ -124,3 +41,12 @@ def conf_get(
global _conf_data global _conf_data
return path_read(_conf_data, path.split(".")) return path_read(_conf_data, path.split("."))
def conf_set(
path : str,
value
):
global _conf_data
path_write(_conf_data, path.split("."), value)

50
source/helpers.py
View file

@ -1,29 +1,3 @@
def convey(x, fs):
y = x
for f in fs:
y = f(y)
return y
def string_coin(
template : str,
arguments : dict
):
result = template
for (key, value, ) in arguments.items():
result = result.replace("{{%s}}" % key, value)
return result
def file_read(
path : str
):
handle = open(path, "r")
content = handle.read()
handle.close()
return content
def log( def log(
messsage : str messsage : str
): ):
@ -42,6 +16,30 @@ def path_read(
return position return position
def path_write(
thing,
steps : List[str],
value
):
steps_first = steps[:-1]
step_last = steps[-1]
position = thing
for step in steps_first:
if (not (step in position)):
position[step] = {}
position = position[step]
position[step_last] = value
def merge(
core,
mantle
):
result = core.copy()
result.update(mantle)
return result
def http_call( def http_call(
request : dict, request : dict,
) -> dict: ) -> dict:

108
source/macros.py
View file

@ -1,6 +1,3 @@
'''
@see https://www.inwx.de/de/help/apidoc/f/ch02.html#account.login
'''
def api_macro_login( def api_macro_login(
environment : str, environment : str,
username : str, username : str,
@ -24,9 +21,6 @@ def api_macro_login(
return response["_accesstoken"] return response["_accesstoken"]
'''
@see https://www.inwx.de/de/help/apidoc/f/ch02.html#account.logout
'''
def api_macro_logout( def api_macro_logout(
environment : str, environment : str,
accesstoken : str accesstoken : str
@ -42,9 +36,6 @@ def api_macro_logout(
return None return None
'''
@see https://www.inwx.de/de/help/apidoc/f/ch02.html#account.info
'''
def api_macro_info( def api_macro_info(
environment : str, environment : str,
username : str, username : str,
@ -63,9 +54,6 @@ def api_macro_info(
return info return info
'''
@see https://www.inwx.de/de/help/apidoc/f/ch02s15.html#nameserver.info
'''
def api_macro_list( def api_macro_list(
environment : str, environment : str,
username : str, username : str,
@ -86,17 +74,12 @@ def api_macro_list(
return info return info
'''
@see https://www.inwx.de/de/help/apidoc/f/ch02s15.html#nameserver.info
@see https://www.inwx.de/de/help/apidoc/f/ch02s15.html#nameserver.createRecord
@see https://www.inwx.de/de/help/apidoc/f/ch02s15.html#nameserver.updateRecord
'''
def api_macro_save( def api_macro_save(
environment : str, environment : str,
username : str, username : str,
password : str, password : str,
domain_base : str, domain : str,
domain_path, name : str,
type_ : str, type_ : str,
content : str content : str
): ):
@ -107,28 +90,12 @@ def api_macro_save(
"nameserver", "nameserver",
"info", "info",
{ {
"domain": domain_base, "domain": domain,
} }
) )
matching = list( matching = list(
filter( filter(
lambda record: ( lambda record: ((record["name"] == (name + "." + domain)) and (record["type"] == type_)),
(
(
(domain_path is None)
and
(record["name"] == domain_base)
)
or
(
(domain_path is not None)
and
(record["name"] == (domain_path + "." + domain_base))
)
)
and
(record["type"] == type_)
),
info["record"] info["record"]
) )
) )
@ -140,8 +107,8 @@ def api_macro_save(
"nameserver", "nameserver",
"createRecord", "createRecord",
{ {
"domain": domain_base, "domain": domain,
"name": domain_path, "name": name,
"type": type_, "type": type_,
"content": content, "content": content,
} }
@ -166,66 +133,3 @@ def api_macro_save(
api_macro_logout(environment, accesstoken) api_macro_logout(environment, accesstoken)
'''
@see https://www.inwx.de/de/help/apidoc/f/ch02s15.html#nameserver.info
@see https://www.inwx.de/de/help/apidoc/f/ch02s15.html#nameserver.deleteRecord
'''
def api_macro_delete(
environment : str,
username : str,
password : str,
domain_base : str,
domain_path,
type_
):
accesstoken = api_macro_login(environment, username, password)
info = api_call(
environment,
accesstoken,
"nameserver",
"info",
{
"domain": domain_base,
}
)
matching = list(
filter(
lambda record: (
(
(
(domain_path is None)
and
(record["name"] == domain_base)
)
or
(
(domain_path is not None)
and
(record["name"] == (domain_path + "." + domain_base))
)
)
and
(
(type_ is None)
or
(record["type"] == type_)
)
),
info["record"]
)
)
for entry in matching:
id_ = entry["id"]
result = api_call(
environment,
accesstoken,
"nameserver",
"deleteRecord",
{
"id": id_,
}
)
api_macro_logout(environment, accesstoken)

280
source/main.py
View file

@ -1,259 +1,149 @@
def main(
def args(
): ):
## args argumentparser = _argparse.ArgumentParser(
argument_parser = _argparse.ArgumentParser(
description = "INWX CLI Frontend" description = "INWX CLI Frontend"
) )
argument_parser.add_argument( argumentparser.add_argument(
"-c", "-c",
"--conf", "--conf",
type = str,
dest = "conf", dest = "conf",
default = _os.path.join(str(_pathlib.Path.home()), ".inwx-conf.json"), default = _os.path.join(str(_pathlib.Path.home()), ".inwx-conf.json"),
metavar = "<conf>", metavar = "<conf>",
help = "path to configuration file", help = "path to configuration file",
) )
argument_parser.add_argument( argumentparser.add_argument(
"-e", "-e",
"--environment", "--environment",
type = str,
dest = "environment", dest = "environment",
metavar = "<environment>", metavar = "<environment>",
default = None, default = None,
help = "environment to use; one of the keys in the 'url' node of the configuration; overwrites the configuration value", help = "environment to use; one of the keys in the 'url' filed of the configuration; overwrites the configuration value",
) )
argument_parser.add_argument( argumentparser.add_argument(
"-u", "-u",
"--username", "--username",
type = str,
dest = "username", dest = "username",
metavar = "<username>", metavar = "<username>",
default = None, default = None,
help = "username; overwrites the configuration value", help = "username; overwrites the configuration value",
) )
argument_parser.add_argument( argumentparser.add_argument(
"-p", "-p",
"--password", "--password",
type = str,
dest = "password", dest = "password",
metavar = "<password>", metavar = "<password>",
default = None, default = None,
help = "password; overwrites the configuration value", help = "password; overwrites the configuration value",
) )
argument_parser.add_argument( '''
argumentparser.add_argument(
"-d", "-d",
"--domain", "--domain",
type = str,
dest = "domain", dest = "domain",
default = None, default = None,
metavar = "<domain>", metavar = "<domain>",
help = "the domain to work with" help = "the domain to work with"
) )
argument_parser.add_argument( '''
"-t", argumentparser.add_argument(
"--type",
type = str,
dest = "type",
default = None,
metavar = "<type>",
help = "the record type (A, AAAA, TXT, …)"
)
argument_parser.add_argument(
"-v",
"--value",
type = str,
dest = "value",
default = None,
metavar = "<value>",
help = "value for the record"
)
argument_parser.add_argument(
"-x", "-x",
"--challenge-prefix", "--challenge-prefix",
type = str,
dest = "challenge_prefix", dest = "challenge_prefix",
metavar = "<challenge-prefix>", metavar = "<challenge-prefix>",
default = "_acme-challenge", default = "_acme-challenge",
help = "which subdomain to use for ACME challanges", help = "which subdomain to use for ACME challanges",
) )
argument_parser.add_argument( argumentparser.add_argument(
"-w", "-w",
"--delay", "--delay",
type = float,
dest = "delay", dest = "delay",
type = float,
default = 60.0, default = 60.0,
metavar = "<delay>", metavar = "<delay>",
help = "seconds to wait at end of certbot auth hook", help = "seconds to wait at end of certbot auth hook",
) )
argument_parser.add_argument( argumentparser.add_argument(
"action",
type = str, type = str,
dest = "action", choices = ["info", "list", "save", "certbot-hook"],
choices = [
"conf-schema",
"info",
"list",
"save",
"delete",
"certbot-hook",
],
metavar = "<action>", metavar = "<action>",
help = string_coin( help = "action to execute",
"action to execute; options:\n{{options}}",
{
"options": convey(
[
{"name": "conf-schema", "requirements": []},
{"name": "info", "requirements": []},
{"name": "list", "requirements": ["<domain>"]},
{"name": "save", "requirements": ["<domain>", "<type>", "<value>"]},
{"name": "delete", "requirements": ["<domain>"]},
{"name": "certbot-hook", "requirements": []},
],
[
lambda x: map(
lambda entry: string_coin(
"{{name}}{{macro_requirements}}",
{
"name": entry["name"],
"macro_requirements": (
""
if (len(entry["requirements"]) <= 0) else
string_coin(
" (requires: {{requirements}})",
{
"requirements": ",".join(entry["requirements"]),
}
)
),
}
),
x
),
" | ".join,
]
)
}
),
) )
args = argument_parser.parse_args() argumentparser.add_argument(
"parameter",
nargs = "*",
type = str,
metavar = "<parameters>",
help = "action specific parameters",
)
arguments = argumentparser.parse_args()
return arguments
## conf
conf_load(args.conf) def main(
):
arguments = args()
## vars conf_load(arguments.conf)
environment = (args.environment or conf_get("environment")) if (not (arguments.environment is None)): conf_set("environment", arguments.environment)
account_username = (args.username or conf_get("account.username")) if (not (arguments.username is None)): conf_set("account.username", arguments.username)
account_password = (args.password or conf_get("account.password")) if (not (arguments.password is None)): conf_set("account.password", arguments.password)
domain_parts = (None if (args.domain is None) else args.domain.split("."))
domain_base = (None if (domain_parts is None) else ".".join(domain_parts[-2:]))
domain_path = (None if ((domain_parts is None) or (len(domain_parts[:-2]) <= 0)) else ".".join(domain_parts[:-2]))
## exec if (arguments.action == "info"):
if (args.action == "conf-schema"): result = api_macro_info(
print(_json.dumps(conf_schema(), indent = "\t")) conf_get("environment"),
elif (args.action == "info"): conf_get("account.username"),
if (account_username is None): conf_get("account.password")
raise ValueError("account username required") )
else: print(_json.dumps(result, indent = "\t"))
if (account_password is None): elif (arguments.action == "list"):
raise ValueError("account password required") domain = arguments.parameter[0]
else: result = api_macro_list(
result = api_macro_info( conf_get("environment"),
environment, conf_get("account.username"),
account_username, conf_get("account.password"),
account_password domain
) )
print(_json.dumps(result, indent = "\t")) print(_json.dumps(result, indent = "\t"))
elif (args.action == "list"): elif (arguments.action == "save"):
if (account_username is None): domain = arguments.parameter[0]
raise ValueError("account username required") name = arguments.parameter[1]
else: type_ = arguments.parameter[2]
if (account_password is None): content = arguments.parameter[3]
raise ValueError("account password required") api_macro_save(
else: conf_get("environment"),
if (args.domain is None): conf_get("account.username"),
raise ValueError("domain required") conf_get("account.password"),
else: domain,
result = api_macro_list( name,
environment, type_,
account_username, content
account_password, )
domain_base # print(_json.dumps(result, indent = "\t"))
) elif (arguments.action == "certbot-hook"):
print(_json.dumps(result, indent = "\t")) domain_full_parts = _os.environ["CERTBOT_DOMAIN"].split(".", 1)
elif (args.action == "save"): domain = domain_full_parts[1]
if (account_username is None): name = (arguments.challenge_prefix + "." + domain_full_parts[0])
raise ValueError("account username required") type_ = "TXT"
else: content = _os.environ["CERTBOT_VALIDATION"]
if (account_password is None): api_macro_save(
raise ValueError("account password required") conf_get("environment"),
else: conf_get("account.username"),
if (args.domain is None): conf_get("account.password"),
raise ValueError("domain required") domain,
else: name,
if (args.type is None): type_,
raise ValueError("type required") content
else: )
if (args.value is None): _time.sleep(arguments.delay)
raise ValueError("value required") # print(_json.dumps(result, indent = "\t"))
else:
api_macro_save(
environment,
account_username,
account_password,
domain_base,
domain_path,
args.type,
args.value
)
elif (args.action == "delete"):
if (account_username is None):
raise ValueError("account username required")
else:
if (account_password is None):
raise ValueError("account password required")
else:
if (args.domain is None):
raise ValueError("domain required")
else:
api_macro_delete(
environment,
account_username,
account_password,
domain_base,
domain_path,
args.type
)
elif (args.action == "certbot-hook"):
if (account_username is None):
raise ValueError("account username required")
else:
if (account_password is None):
raise ValueError("account password required")
else:
domain_full_parts = _os.environ["CERTBOT_DOMAIN"].split(".")
domain_base = ".".join(domain_full_parts[-2:])
domain_path_stripped = ".".join(domain_full_parts[:-2])
domain_path = (args.challenge_prefix + "." + domain_path_stripped)
type_ = "TXT"
content = _os.environ["CERTBOT_VALIDATION"]
api_macro_save(
environment,
account_username,
account_password,
domain_base,
domain_path,
type_,
content
)
_time.sleep(args.delay)
# print(_json.dumps(result, indent = "\t"))
else: else:
log("unhandled action '%s'" % (args.action, )) log("unhandled action '%s'" % (arguments.action, ))
try: try:
main() main()
except ValueError as error: except ValueError as error:
_sys.stderr.write("-- %s\n" % str(error)) _sys.stderr.write(str(error) + "\n")

4
tools/build
View file

@ -1,4 +0,0 @@
#!/usr/bin/env sh
make -f makefile