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

[ini]

Browse source
This commit is contained in:
fenris 2026-03-06 08:39:17 +01:00
commit 189682b069
11 changed files with 845 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.gitignore vendored Normal file
View file

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

23
makefile Normal file
View file

@ -0,0 +1,23 @@
dir_source := source
dir_build := build
cmd_md := mkdir -p
cmd_log := echo "--"
cmd_cat := cat
cmd_chmod := chmod
_default: ${dir_build}/inwx
.PHONY: _default
${dir_build}/inwx: \
${dir_source}/head.py \
${dir_source}/helpers.py \
${dir_source}/conf.py \
${dir_source}/core.py \
${dir_source}/macros.py \
${dir_source}/main.py
@ ${cmd_log} "building …"
@ ${cmd_md} ${dir_build}
@ ${cmd_cat} $^ > $@
@ ${cmd_chmod} +x $@

60
readme.md Normal file
View file

@ -0,0 +1,60 @@
# Description
A simple CLI client for the API of [INWX](inwx.de)
# Usage
## Credentials
For most API calls it is necessary to provide login information. There are two ways to do this:
### Via command line arguments
- `--username` for specifying the username of the account
- `--password` for specifying the username of the account
### Via configuration file
- the location of a configuration file can be specified via `--conf`
- the default location is `~/.inwx-conf.json`
- a minial configuration file for specifying the credentials would look as follows:
```
{
"account": {
"username": "___",
"password": "___"
}
}
```
## Commands
### `info`
- synopsis: `inwx info`
- description: for listing the records of a domain
### `list`
- synopsis: `inwx list <domain>`
- description: for listing the records of a domain
### `save`
- synopsis: `inwx save <domain> <name> <type> <content>`
- description: for creating or updating a records of a domain
- example: `inwx save example.org dat TXT 'foo bar'`
### `certbot-hook`
- synopsis: `inwx certbot-hook`
- description: for executing the DNS certbot challenge; will read the environment variables `CERTBOT_DOMAIN` and `CERTBOT_VALIDATION` to store a `TXT` record

126
source/conf.py Normal file
View file

@ -0,0 +1,126 @@
_conf_data = None
def conf_schema(
):
return {
"type": "object",
"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,
"required": [
],
}
def conf_load(
path : str
):
global _conf_data
conf_data_raw = (
_json.loads(file_read(path))
if _os.path.exists(path) else
{}
)
for pair in conf_data_raw.get("url", {}).items():
if ("host" in pair[1]):
pass
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(
path : str
):
global _conf_data
return path_read(_conf_data, path.split("."))

44
source/core.py Normal file
View file

@ -0,0 +1,44 @@
def api_call(
environment : str,
accesstoken : str,
category : str,
action : str,
data,
):
url = conf_get("url." + environment)
# input_["lang"] = "de"
request_headers = {
"Content-Type": "application/json",
}
if (accesstoken is not None):
request_headers["Cookie"] = ("domrobot=%s" % (accesstoken, ))
else:
pass
request_data_decoded = {
"method": (category + "." + action),
"params": data,
}
request = {
"url": url,
"method": "POST",
"headers": request_headers,
"data": _json.dumps(request_data_decoded),
}
# log("[>>] %s" % _json.dumps(request, indent = "\t"))
response = http_call(request)
# log("[<<] %s" % _json.dumps(response, indent = "\t"))
if (not (response["status"] == 200)):
raise ValueError("API call failed with status %u: %s" % (response["status"], response["data"], ))
else:
output_data_decoded = _json.loads(response["data"])
result = (output_data_decoded["resData"] if ("resData" in output_data_decoded) else {})
if ("Set-Cookie" in response["headers"]):
result["_accesstoken"] = response["headers"]["Set-Cookie"].split("; ")[0].split("=")[1]
else:
pass
if (output_data_decoded["code"] == 2002):
raise ValueError("wrong use: %s" % str(output_data_decoded))
else:
return result

12
source/head.py Normal file
View file

@ -0,0 +1,12 @@
#!/usr/bin/env python3
from typing import List
import os as _os
import sys as _sys
import json as _json
import http.client as _http_client
import argparse as _argparse
import pathlib as _pathlib
import time as _time

68
source/helpers.py Normal file
View file

@ -0,0 +1,68 @@
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(
messsage : str
):
_sys.stderr.write("-- %s\n" % messsage)
def path_read(
thing,
steps : List[str]
):
position = thing
for step in steps:
if (not (step in position)):
raise ValueError("missing key '%s'" % ".".join(steps))
position = position[step]
return position
def http_call(
request : dict,
) -> dict:
connection = (
{
"http": (lambda: _http_client.HTTPConnection(request["url"]["host"], request["url"]["port"])),
"https": (lambda: _http_client.HTTPSConnection(request["url"]["host"], request["url"]["port"])),
}[request["url"]["scheme"]]
)()
connection.request(
request["method"],
("/" + request["url"]["path"]),
request["data"],
request["headers"]
)
response_ = connection.getresponse()
response = {
"status": response_.status,
"headers": dict(response_.getheaders()),
"data": response_.read(),
}
return response

231
source/macros.py Normal file
View file

@ -0,0 +1,231 @@
'''
@see https://www.inwx.de/de/help/apidoc/f/ch02.html#account.login
'''
def api_macro_login(
environment : str,
username : str,
password : str
):
if ((username is None) or (password is None)):
raise ValueError("username or password not given")
else:
response = (
api_call(
environment,
None,
"account",
"login",
{
"user": username,
"pass": password,
}
)
)
return response["_accesstoken"]
'''
@see https://www.inwx.de/de/help/apidoc/f/ch02.html#account.logout
'''
def api_macro_logout(
environment : str,
accesstoken : str
):
response = api_call(
environment,
accesstoken,
"account",
"logout",
{
}
)
return None
'''
@see https://www.inwx.de/de/help/apidoc/f/ch02.html#account.info
'''
def api_macro_info(
environment : str,
username : str,
password : str
):
accesstoken = api_macro_login(environment, username, password)
info = api_call(
environment,
accesstoken,
"account",
"info",
{
}
)
api_macro_logout(environment, accesstoken)
return info
'''
@see https://www.inwx.de/de/help/apidoc/f/ch02s15.html#nameserver.info
'''
def api_macro_list(
environment : str,
username : str,
password : str,
domain : str
):
accesstoken = api_macro_login(environment, username, password)
info = api_call(
environment,
accesstoken,
"nameserver",
"info",
{
"domain": domain,
}
)
api_macro_logout(environment, accesstoken)
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(
environment : str,
username : str,
password : str,
domain_base : str,
domain_path,
type_ : str,
content : str
):
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
(record["type"] == type_)
),
info["record"]
)
)
count = len(matching)
if (count == 0):
result = api_call(
environment,
accesstoken,
"nameserver",
"createRecord",
{
"domain": domain_base,
"name": domain_path,
"type": type_,
"content": content,
}
)
id_ = result["id"]
log("created record %s" % id_)
elif (count == 1):
id_ = matching[0]["id"]
result = api_call(
environment,
accesstoken,
"nameserver",
"updateRecord",
{
"id": id_,
"content": content,
}
)
log("updated record %s" % id_)
else:
log("found multiple records with this name and type")
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)

259
source/main.py Normal file
View file

@ -0,0 +1,259 @@
def main(
):
## args
argument_parser = _argparse.ArgumentParser(
description = "INWX CLI Frontend"
)
argument_parser.add_argument(
"-c",
"--conf",
type = str,
dest = "conf",
default = _os.path.join(str(_pathlib.Path.home()), ".inwx-conf.json"),
metavar = "<conf>",
help = "path to configuration file",
)
argument_parser.add_argument(
"-e",
"--environment",
type = str,
dest = "environment",
metavar = "<environment>",
default = None,
help = "environment to use; one of the keys in the 'url' node of the configuration; overwrites the configuration value",
)
argument_parser.add_argument(
"-u",
"--username",
type = str,
dest = "username",
metavar = "<username>",
default = None,
help = "username; overwrites the configuration value",
)
argument_parser.add_argument(
"-p",
"--password",
type = str,
dest = "password",
metavar = "<password>",
default = None,
help = "password; overwrites the configuration value",
)
argument_parser.add_argument(
"-d",
"--domain",
type = str,
dest = "domain",
default = None,
metavar = "<domain>",
help = "the domain to work with"
)
argument_parser.add_argument(
"-t",
"--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",
"--challenge-prefix",
type = str,
dest = "challenge_prefix",
metavar = "<challenge-prefix>",
default = "_acme-challenge",
help = "which subdomain to use for ACME challanges",
)
argument_parser.add_argument(
"-w",
"--delay",
type = float,
dest = "delay",
default = 60.0,
metavar = "<delay>",
help = "seconds to wait at end of certbot auth hook",
)
argument_parser.add_argument(
type = str,
dest = "action",
choices = [
"conf-schema",
"info",
"list",
"save",
"delete",
"certbot-hook",
],
metavar = "<action>",
help = string_coin(
"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()
## conf
conf_load(args.conf)
## vars
environment = (args.environment or conf_get("environment"))
account_username = (args.username or conf_get("account.username"))
account_password = (args.password or conf_get("account.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 (args.action == "conf-schema"):
print(_json.dumps(conf_schema(), indent = "\t"))
elif (args.action == "info"):
if (account_username is None):
raise ValueError("account username required")
else:
if (account_password is None):
raise ValueError("account password required")
else:
result = api_macro_info(
environment,
account_username,
account_password
)
print(_json.dumps(result, indent = "\t"))
elif (args.action == "list"):
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:
result = api_macro_list(
environment,
account_username,
account_password,
domain_base
)
print(_json.dumps(result, indent = "\t"))
elif (args.action == "save"):
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:
if (args.type is None):
raise ValueError("type required")
else:
if (args.value is None):
raise ValueError("value required")
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:
log("unhandled action '%s'" % (args.action, ))
try:
main()
except ValueError as error:
_sys.stderr.write("-- %s\n" % str(error))

4
tools/build Executable file
View file

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

16
tools/install Executable file
View file

@ -0,0 +1,16 @@
#!/usr/bin/env sh
## consts
dir_build="build"
## args
if [ $# -ge 1 ] ; then dir_target=$1 ; else dir_target=/usr/local/bin ; fi
## exec
cp ${dir_build}/inwx ${dir_target}/inwx