misc/ansible-base
0
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

[fix] role:tlscert_acme_inwx

Browse source
This commit is contained in:
Christian Fraß 2023-11-20 16:01:49 +01:00
parent d4cd307ebf
commit fbd3e6d6cc
4 changed files with 508 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
ansible/roles/tlscert_acme_inwx/defaults/main.json
View file

@ -1,5 +1,6 @@
{
"var_tlscert_acme_inwx_acme_account_email": "REPLACE_ME",
"var_tlscert_acme_inwx_letsencrypt_account_email": "REPLACE_ME",
"var_tlscert_acme_inwx_letsencrypt_account_key_path": "/etc/letsencrypt/key",
"var_tlscert_acme_inwx_inwx_account_username": "REPLACE_ME",
"var_tlscert_acme_inwx_inwx_account_password": "REPLACE_ME",
"var_tlscert_acme_inwx_domain_base": "example.org",

460
ansible/roles/tlscert_acme_inwx/files/inwx Executable file
View file

@ -0,0 +1,460 @@
#!/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
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 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(
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
_conf_data = {
"url": {
"test": {
"scheme": "https",
"host": "api.ote.domrobot.com",
"port": 443,
"path": "jsonrpc/"
},
"production": {
"scheme": "https",
"host": "api.domrobot.com",
"port": 443,
"path": "jsonrpc/"
}
},
"environment": "production",
"account": {
"username": None,
"password": None
}
}
def conf_load(
path : str
):
global _conf_data
if (not _os.path.exists(path)):
pass
else:
handle = open(path, "r")
content = handle.read()
handle.close()
data = _json.loads(content)
_conf_data = merge(_conf_data, data)
def conf_get(
path : str
):
global _conf_data
return path_read(_conf_data, path.split("."))
def conf_set(
path : str,
value
):
global _conf_data
path_write(_conf_data, path.split("."), value)
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
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"]
def api_macro_logout(
environment : str,
accesstoken : str
):
response = api_call(
environment,
accesstoken,
"account",
"logout",
{
}
)
return None
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
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
def api_macro_save(
environment : str,
username : str,
password : str,
domain : str,
name : str,
type_ : str,
content : str
):
accesstoken = api_macro_login(environment, username, password)
info = api_call(
environment,
accesstoken,
"nameserver",
"info",
{
"domain": domain,
}
)
matching = list(
filter(
lambda record: ((record["name"] == (name + "." + domain)) and (record["type"] == type_)),
info["record"]
)
)
count = len(matching)
if (count == 0):
result = api_call(
environment,
accesstoken,
"nameserver",
"createRecord",
{
"domain": domain,
"name": name,
"type": type_,
"content": content,
}
)
id_ = result["id"]
log("created record %u" % id_)
elif (count == 1):
id_ = matching[0]["id"]
result = api_call(
environment,
accesstoken,
"nameserver",
"updateRecord",
{
"id": id_,
"content": content,
}
)
log("updated record %u" % id_)
else:
log("found multiple records with this name and type")
api_macro_logout(environment, accesstoken)
def args(
):
argumentparser = _argparse.ArgumentParser(
description = "INWX CLI Frontend"
)
argumentparser.add_argument(
"-c",
"--conf",
dest = "conf",
default = _os.path.join(str(_pathlib.Path.home()), ".inwx-conf.json"),
metavar = "<conf>",
help = "path to configuration file",
)
argumentparser.add_argument(
"-e",
"--environment",
dest = "environment",
metavar = "<environment>",
default = None,
help = "environment to use; one of the keys in the 'url' filed of the configuration; overwrites the configuration value",
)
argumentparser.add_argument(
"-u",
"--username",
dest = "username",
metavar = "<username>",
default = None,
help = "username; overwrites the configuration value",
)
argumentparser.add_argument(
"-p",
"--password",
dest = "password",
metavar = "<password>",
default = None,
help = "password; overwrites the configuration value",
)
'''
argumentparser.add_argument(
"-d",
"--domain",
dest = "domain",
default = None,
metavar = "<domain>",
help = "the domain to work with"
)
'''
argumentparser.add_argument(
"-x",
"--challenge-prefix",
dest = "challenge_prefix",
metavar = "<challenge-prefix>",
default = "_acme-challenge",
help = "which subdomain to use for ACME challanges",
)
argumentparser.add_argument(
"-w",
"--delay",
dest = "delay",
type = float,
default = 60.0,
metavar = "<delay>",
help = "seconds to wait at end of certbot auth hook",
)
argumentparser.add_argument(
"action",
type = str,
choices = ["info", "list", "save", "certbot-hook"],
metavar = "<action>",
help = "action to execute",
)
argumentparser.add_argument(
"parameter",
nargs = "*",
type = str,
metavar = "<parameters>",
help = "action specific parameters",
)
arguments = argumentparser.parse_args()
return arguments
def main(
):
arguments = args()
conf_load(arguments.conf)
if (not (arguments.environment is None)): conf_set("environment", arguments.environment)
if (not (arguments.username is None)): conf_set("account.username", arguments.username)
if (not (arguments.password is None)): conf_set("account.password", arguments.password)
if (arguments.action == "info"):
result = api_macro_info(
conf_get("environment"),
conf_get("account.username"),
conf_get("account.password")
)
print(_json.dumps(result, indent = "\t"))
elif (arguments.action == "list"):
domain = arguments.parameter[0]
result = api_macro_list(
conf_get("environment"),
conf_get("account.username"),
conf_get("account.password"),
domain
)
print(_json.dumps(result, indent = "\t"))
elif (arguments.action == "save"):
domain = arguments.parameter[0]
name = arguments.parameter[1]
type_ = arguments.parameter[2]
content = arguments.parameter[3]
api_macro_save(
conf_get("environment"),
conf_get("account.username"),
conf_get("account.password"),
domain,
name,
type_,
content
)
# print(_json.dumps(result, indent = "\t"))
elif (arguments.action == "certbot-hook"):
domain_full_parts = _os.environ["CERTBOT_DOMAIN"].split(".")
account = ".".join(domain_full_parts[-2:])
concern = ".".join(domain_full_parts[:-2])
domain = account
name = (arguments.challenge_prefix + "." + concern)
type_ = "TXT"
content = _os.environ["CERTBOT_VALIDATION"]
api_macro_save(
conf_get("environment"),
conf_get("account.username"),
conf_get("account.password"),
domain,
name,
type_,
content
)
_time.sleep(arguments.delay)
# print(_json.dumps(result, indent = "\t"))
else:
log("unhandled action '%s'" % (arguments.action, ))
try:
main()
except ValueError as error:
_sys.stderr.write(str(error) + "\n")

6
ansible/roles/tlscert_acme_inwx/info.md
View file

@ -1,3 +1,9 @@
## Verweise
- [Digital Ocean | How To Acquire a Let's Encrypt Certificate Using Ansible](https://www.digitalocean.com/community/tutorials/how-to-acquire-a-let-s-encrypt-certificate-using-ansible-on-ubuntu-18-04)
- [INWX | API-Informationen](https://www.inwx.de/de/offer/api)
## ToDo
- inwx-Skript von richtiger Quelle holen

50
ansible/roles/tlscert_acme_inwx/tasks/main.json
View file

@ -4,13 +4,12 @@
"become": true,
"ansible.builtin.apt": {
"pkg": [
"openssl",
"certbot"
"openssl"
]
}
},
{
"name": "csr | setup private key directory",
"name": "setup directories | keys",
"become": true,
"ansible.builtin.file": {
"state": "directory",
@ -18,20 +17,44 @@
}
},
{
"name": "csr | generate private key",
"name": "setup directories | certs",
"become": true,
"community.crypto.openssl_privatekey": {
"path": "{{var_tlscert_acme_inwx_ssl_directory}}/private/{{var_tlscert_acme_inwx_domain_path}}.{{var_tlscert_acme_inwx_domain_base}}.pem"
"ansible.builtin.file": {
"state": "directory",
"path": "{{var_tlscert_acme_inwx_ssl_directory}}/certs"
}
},
{
"name": "csr | setup csr directory",
"name": "setup directories | csr",
"become": true,
"ansible.builtin.file": {
"state": "directory",
"path": "{{var_tlscert_acme_inwx_ssl_directory}}/csr"
}
},
{
"name": "setup directories | fullchains",
"become": true,
"ansible.builtin.file": {
"state": "directory",
"path": "{{var_tlscert_acme_inwx_ssl_directory}}/fullchains"
}
},
{
"name": "setup directories | Let's Encrypt account key",
"become": true,
"ansible.builtin.file": {
"state": "directory",
"path": "{{var_tlscert_acme_inwx_letsencrypt_account_key_path | dirname}}"
}
},
{
"name": "csr | generate private key",
"become": true,
"community.crypto.openssl_privatekey": {
"path": "{{var_tlscert_acme_inwx_ssl_directory}}/private/{{var_tlscert_acme_inwx_domain_path}}.{{var_tlscert_acme_inwx_domain_base}}.pem"
}
},
{
"name": "csr | execute",
"become": true,
@ -41,6 +64,13 @@
"path": "{{var_tlscert_acme_inwx_ssl_directory}}/csr/{{var_tlscert_acme_inwx_domain_path}}.{{var_tlscert_acme_inwx_domain_base}}.pem"
}
},
{
"name": "acme | generate account key",
"become": true,
"ansible.builtin.shell": {
"cmd": "test -f {{var_tlscert_acme_inwx_letsencrypt_account_key_path}} || openssl genrsa 4096 > {{var_tlscert_acme_inwx_letsencrypt_account_key_path}}"
}
},
{
"name": "acme | init",
"become": true,
@ -48,7 +78,7 @@
"acme_version": 2,
"acme_directory": "https://acme-v02.api.letsencrypt.org/directory",
"account_email": "{{var_tlscert_acme_inwx_acme_account_email}}",
"account_key_src": "{{var_tlscert_acme_inwx_ssl_directory}}/private/{{var_tlscert_acme_inwx_domain_path}}.{{var_tlscert_acme_inwx_domain_base}}.pem",
"account_key_src": "{{var_tlscert_acme_inwx_letsencrypt_account_key_path}}",
"terms_agreed": true,
"csr": "{{var_tlscert_acme_inwx_ssl_directory}}/csr/{{var_tlscert_acme_inwx_domain_path}}.{{var_tlscert_acme_inwx_domain_base}}.pem",
"challenge": "dns-01",
@ -61,7 +91,7 @@
"name": "dns challenge | place script",
"become": true,
"ansible.builtin.copy": {
"src": "/usr/local/bin/inwx",
"src": "inwx",
"dest": "/usr/local/bin/inwx",
"mode": "a+x"
}
@ -85,7 +115,7 @@
"acme_version": 2,
"acme_directory": "https://acme-v02.api.letsencrypt.org/directory",
"account_email": "{{var_tlscert_acme_inwx_acme_account_email}}",
"account_key_src": "{{var_tlscert_acme_inwx_ssl_directory}}/private/{{var_tlscert_acme_inwx_domain_path}}.{{var_tlscert_acme_inwx_domain_base}}.pem",
"account_key_src": "{{var_tlscert_acme_inwx_letsencrypt_account_key_path}}",
"terms_agreed": true,
"csr": "{{var_tlscert_acme_inwx_ssl_directory}}/csr/{{var_tlscert_acme_inwx_domain_path}}.{{var_tlscert_acme_inwx_domain_base}}.pem",
"challenge": "dns-01",