Separated functions of the Python script.

vault_cert_renew.py

@@ -1,6 +1,5 @@
 #!/usr/bin/python3
 import argparse
-import getpass
 import json
 import os
 import ssl
@@ -10,38 +9,32 @@ import tempfile
 import urllib.error
 import urllib.request
 
-parser = argparse.ArgumentParser(description="Renew the client certificate from Vault with AppRole.")
-parser.add_argument(
-    "--unwrap",
-    action="store_true",
-    help="read a response wrapping token and store the unwrapped secret ID",
-)
-parser.add_argument(
-    "--login-test",
-    action="store_true",
-    help="log in with AppRole, print the token policies and lifetime, and exit",
-)
+parser = argparse.ArgumentParser(description="Renew a certificate from Vault with AppRole.")
+parser.add_argument("config", help="path to the JSON configuration file of the instance")
 args = parser.parse_args()
 
+with open(args.config) as file:
+    config = json.load(file)
+
 
 def setting(name, default=None):
-    value = os.environ.get(name, default)
+    value = config.get(name, default)
     if value is None:
-        sys.exit(f"{name} is not set")
+        sys.exit(f"{name} is not set in {args.config}")
     return value
 
 
-vault_addr = setting("VAULT_ADDR").rstrip("/")
-context = ssl.create_default_context(cafile=os.environ.get("VAULT_CACERT") or None)
+vault_addr = setting("vault_addr").rstrip("/")
+context = ssl.create_default_context(cafile=config.get("vault_cacert") or None)
 
 
-def api(path, body=None, token=None):
+def api(path, body, token=None):
     headers = {"Content-Type": "application/json"}
     if token:
         headers["X-Vault-Token"] = token
     request = urllib.request.Request(
         f"{vault_addr}/v1/{path}",
-        data=json.dumps(body or {}).encode(),
+        data=json.dumps(body).encode(),
         headers=headers,
         method="POST",
     )
@@ -52,41 +45,12 @@ def api(path, body=None, token=None):
         sys.exit(f"Vault returned {error.code} for {path}: {error.read().decode()}")
 
 
-role_id_file = setting("ROLE_ID_FILE", "/etc/vault-cert/role_id")
-secret_id_file = setting("SECRET_ID_FILE", "/etc/vault-cert/secret_id")
-
-
-def login():
-    with open(role_id_file) as file:
-        role_id = file.read().strip()
-    with open(secret_id_file) as file:
-        secret_id = file.read().strip()
-    return api("auth/approle/login", {"role_id": role_id, "secret_id": secret_id})["auth"]
-
-
-# Store the secret ID that is wrapped in a single use response wrapping token.
-if args.unwrap:
-    wrapping_token = getpass.getpass("Wrapping token: ")
-    secret_id = api("sys/wrapping/unwrap", token=wrapping_token)["data"]["secret_id"]
-    descriptor = os.open(secret_id_file, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
-    os.fchmod(descriptor, 0o600)
-    with os.fdopen(descriptor, "w") as file:
-        file.write(secret_id)
-    print(f"The secret ID was written to {secret_id_file}")
-    sys.exit(0)
-
-# Check the AppRole credentials without issuing a certificate.
-if args.login_test:
-    auth = login()
-    print(f"Login succeeded: token type {auth['token_type']}, policies {auth['token_policies']}, valid for {auth['lease_duration']} seconds")
-    sys.exit(0)
-
-cert_dir = setting("CERT_DIR", "/var/lib/vault-cert")
-common_name = setting("CERT_CN")
-sign_path = setting("SIGN_PATH", "pki_int/sign/wif-client")
-renew_before_days = int(setting("RENEW_BEFORE_DAYS", "30"))
-cert_ttl = setting("CERT_TTL", "8760h")
-hook = os.environ.get("HOOK", "")
+cert_dir = setting("cert_dir", "/var/lib/vault-cert")
+common_name = setting("cert_cn")
+sign_path = setting("sign_path", "pki_int/sign/wif-client")
+renew_before_days = int(setting("renew_before_days", 30))
+cert_ttl = setting("cert_ttl", "8760h")
+hook = config.get("hook", "")
 
 cert = os.path.join(cert_dir, "client_cert.pem")
 key = os.path.join(cert_dir, "client_key.pem")
@@ -108,7 +72,8 @@ with tempfile.TemporaryDirectory(dir=cert_dir, prefix=".renew.") as workdir:
     csr_file = os.path.join(workdir, "client.csr")
 
     print("Logging in to Vault with AppRole")
-    token = login()["client_token"]
+    login = {"role_id": setting("role_id"), "secret_id": setting("secret_id")}
+    token = api("auth/approle/login", login)["auth"]["client_token"]
 
     print(f"Generating the key and the CSR for {common_name}")
     subprocess.run(