diff --git a/nyacme/config.py b/nyacme/config.py index ced6206..21b8779 100644 --- a/nyacme/config.py +++ b/nyacme/config.py @@ -1,8 +1,8 @@ import logging import os import sys -from typing import Any, Optional from pathlib import Path +from typing import Any import tomllib @@ -40,15 +40,30 @@ def config_parse_dict(raw_conf: dict[str, Any], key: str) -> dict[str, str]: sys.exit(1) for k, v in raw_conf[key].items(): - assert isinstance(k, str), f'"{k}" is not a string' - assert isinstance(v, str), f'"{k}" value "{v}" is not a string' + if not isinstance(k, str): + raise TypeError(f'"{k}" is not a string') + if not isinstance(v, str): + raise TypeError(f'"{k}" value "{v}" is not a string') result: dict[str, str] = raw_conf[key] return result +def config_parse_list(raw_conf: dict[str, Any], key: str) -> list[str]: + result: list[str] = [] -def read_config(path: Optional[str]) -> Config: + if not isinstance(raw_conf[key], list): + raise TypeError(f'"{key}" must be a list, not {type(raw_conf[key]).__name__}') + + for item in raw_conf[key]: + if not isinstance(item, str): + raise TypeError(f'"{key}" list item must be a string') + result.append(item) + + return result + + +def read_config(path: str | None) -> Config: if not path: # should be here only when running from hook path = os.getenv('NYACME_CONFIG', '/etc/nyacme.toml') @@ -67,10 +82,7 @@ def read_config(path: Optional[str]) -> Config: post_acquire = [] if 'post_acquire' in raw_conf: - assert isinstance(raw_conf['post_acquire'], list), 'post_acquire is not a list' - for cmd in raw_conf['post_acquire']: - assert isinstance(cmd, str), 'post_acquire item has to be a string' - post_acquire.append(cmd) + post_acquire = config_parse_list(raw_conf, 'post_acquire') c.post_acquire = post_acquire @@ -83,9 +95,6 @@ def read_config(path: Optional[str]) -> Config: log.error('missing "certificates"') sys.exit(1) - c.certificates = [] - for cert in raw_conf['certificates']: - assert isinstance(cert, str), 'certificate should be a string' - c.certificates.append(cert) + c.certificates = config_parse_list(raw_conf, 'certificates') return c diff --git a/nyacme/handlers/he.py b/nyacme/handlers/he.py index cf2552e..8c40816 100644 --- a/nyacme/handlers/he.py +++ b/nyacme/handlers/he.py @@ -29,7 +29,7 @@ class HEHandler(Handler): except urllib.error.HTTPError as ex: self.log.error('cannot set the record %s: %s', record_name, ex) res = ex.fp.read().decode('utf-8') - raise Exception(res) + raise Exception(res) from ex def create(self, record_name: str, record_value: str) -> None: self.log.info('creating %s with value %s', record_name, record_value) diff --git a/nyacme/handlers/hetzner.py b/nyacme/handlers/hetzner.py index 78172f6..ec44e45 100644 --- a/nyacme/handlers/hetzner.py +++ b/nyacme/handlers/hetzner.py @@ -1,6 +1,6 @@ import json import urllib.request -from typing import Any, Optional +from typing import Any from ..config import Config from .base import Handler @@ -22,7 +22,7 @@ class HetznerHandler(Handler): self.nameservers = zone['ns'] break - def fetch(self, url: str, data: Optional[Any] = None, **kwargs: str) -> Any: + def fetch(self, url: str, data: Any | None = None, **kwargs: str) -> Any: req = urllib.request.Request('https://dns.hetzner.com/api/v1' + url) req.add_header('Auth-API-Token', self.secret) method = 'GET' @@ -39,9 +39,9 @@ class HetznerHandler(Handler): self.log.error('cannot %s %s: %s', req.method, url, ex) res = ex.fp.read().decode('utf-8') try: - raise Exception(json.loads(res)['error']) + raise Exception(json.loads(res)['error']) from None except json.JSONDecodeError: - raise Exception(res) + raise Exception(res) from None def create(self, record_name: str, record_value: str) -> None: self.remove(record_name) diff --git a/nyacme/handlers/porkbun.py b/nyacme/handlers/porkbun.py index a7c26fa..e061537 100644 --- a/nyacme/handlers/porkbun.py +++ b/nyacme/handlers/porkbun.py @@ -17,9 +17,11 @@ class PorkbunHandler(Handler): self.nameservers = self.fetch(f'/domain/getNs/{self.zone}')['ns'] - def fetch(self, url: str, data: dict[str, Any] = {}) -> Any: + def fetch(self, url: str, data: dict[str, Any] | None = None) -> Any: req = urllib.request.Request('https://api.porkbun.com/api/json/v3' + url) + if data is None: + data = {} data['apikey'] = self.apikey data['secretapikey'] = self.secretapikey req.data = json.dumps(data).encode('utf-8') @@ -27,17 +29,17 @@ class PorkbunHandler(Handler): req.method = 'POST' try: with urllib.request.urlopen(req) as f: - data = json.load(f) - if data['status'] != 'SUCCESS': - raise Exception(data['message']) - return data + res = json.load(f) + if res['status'] != 'SUCCESS': + raise Exception(res['message']) + return res except urllib.error.HTTPError as ex: self.log.error('cannot %s %s: %s', req.method, url, ex) res = ex.fp.read().decode('utf-8') try: raise Exception(json.loads(res)['message']) except json.JSONDecodeError: - raise Exception(res) + raise Exception(res) from None def create(self, record_name: str, record_value: str) -> None: self.remove(record_name)