karolyi
/
daneupdate
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Comparing existing records with required

This commit is contained in:
László Károlyi 2024-04-05 17:01:46 +02:00
parent 4fe359a3c1
commit 47d8f6212c
Signed by: karolyi
GPG Key ID: 2DCAF25E55735BFE
9 changed files with 198 additions and 106 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
credits.txt Normal file
View File

@ -0,0 +1,2 @@
- https://github.com/shuque/tlsa_rdata for being a huge help in generating the TLSA record variants
- https://letsdns.org/ another TLSA record generator, not intertwined with certbot

10
src/daneupdate/handlers/base.py
View File

@ -15,8 +15,8 @@ from certbot._internal.storage import renewal_conf_files
from certbot.crypto_util import verify_renewable_cert
from typing_extensions import Self
from ..utils.cert import DaneCert
from ..utils.config import Configuration
from ..utils.config_types.cert import DaneRenewableCert
_LOGGER = getLogger(name=__name__)
@ -24,8 +24,8 @@ _LOGGER = getLogger(name=__name__)
class _ParsedCertsResult(NamedTuple):
'Result of a `_get_parsed_certs()` call.'
# A `MappingProxyType` is basically a read-only `dict`
# config path -> DaneRenewableCert
parsed_certs: MappingProxyType[str, DaneRenewableCert]
# config path -> DaneCert
parsed_certs: MappingProxyType[str, DaneCert]
# lineage name -> config path
lineages: MappingProxyType[str, str]
parse_failures: tuple[str, ...]
@ -62,12 +62,12 @@ class HandlerBase(object, metaclass=ABCMeta):
See `certbot._internal.cert_manager.certificates()`.
"""
parsed_certs = dict[str, DaneRenewableCert]()
parsed_certs = dict[str, DaneCert]()
parse_failures = tuple[str, ...]()
lineages = dict[str, str]()
for renewal_file in renewal_conf_files(config=self._certbot_config):
try:
renewal_candidate = DaneRenewableCert.from_data(
renewal_candidate = DaneCert.from_configdata(
config_filename=renewal_file,
cli_config=self._certbot_config)
verify_renewable_cert(renewable_cert=renewal_candidate)

18
src/daneupdate/handlers/renew.py
View File

@ -23,7 +23,7 @@ from certbot.interfaces import Authenticator, Installer
from certbot.util import Key
from ktools.pathlib import Path
from ..utils.config_types.cert import DaneRenewableCert
from ..utils.cert import DaneCert
from ..utils.tlsa.updater import TlsaUpdater
from .base import HandlerBase
@ -39,14 +39,11 @@ _LOGGER = getLogger(name=__name__)
class _CertbotLogicEmulator(object):
"""
Emulating certbot renewal operations, for **one**
`DaneRenewableCert`.
"""
'Emulating certbot renewal operations, for **one** `DaneCert`.'
def __init__(
self, certbot_config: NamespaceConfig, plugins: PluginsRegistry,
cert: DaneRenewableCert):
cert: DaneCert):
self._certbot_config = certbot_config
self.cert = cert
self._plugins = plugins
@ -186,7 +183,7 @@ class RenewHandler(HandlerBase):
return args
def _put_in_place_when_ttl_passed(
self, certconfig_path: str, cert: DaneRenewableCert) -> list[str]:
self, certconfig_path: str, cert: DaneCert) -> list[str]:
"""
If the cert `mtime` has passed the TLSA record's TTL, put it in
place.
@ -208,8 +205,7 @@ class RenewHandler(HandlerBase):
return certbot_emu.run_installer_logic()
def _renew_one_adopted(
self, certconfig_path: str, cert: DaneRenewableCert
) -> list[str]:
self, certconfig_path: str, cert: DaneCert) -> list[str]:
"""
Handle renewal/update of an adopted certificate.
:returns: The list of renewed domains in case a previously
@ -235,14 +231,14 @@ class RenewHandler(HandlerBase):
return []
def _renew_one_nonadopted(
self, cert: DaneRenewableCert) -> tuple[list[str], list[str]]:
self, cert: DaneCert) -> tuple[list[str], list[str]]:
'Renew a non-adopted certificate by running certbot\'s logic.'
config = deepcopy(self._certbot_config)
config.certname = cert.lineagename
return handle_renewal_request(config=config)
def _start_renewing(
self, to_renew: dict[str, DaneRenewableCert],
self, to_renew: dict[str, DaneCert],
already_adopted: frozenset[str]) -> int:
'Start renewing the chosen certificates.'
renewed_domains = list[str]()

107
src/daneupdate/utils/config_types/cert.py → src/daneupdate/utils/cert.py
View File

@ -1,6 +1,7 @@
from __future__ import annotations
from functools import cached_property
from functools import cached_property, lru_cache
from hashlib import sha256, sha512
from logging import getLogger
from typing import Literal, Mapping
@ -11,14 +12,39 @@ from certbot.compat.filesystem import (
compute_private_key_mode, copy_ownership_and_apply_mode)
from certbot.configuration import NamespaceConfig
from certbot.util import Key
from cryptography.hazmat.primitives._serialization import (
Encoding, PublicFormat)
from cryptography.x509.base import Certificate, load_pem_x509_certificate
from ktools.pathlib import Path
from .config_types.common import TlsaMatchingType, TlsaSelector, TlsaUsage
from .tlsa.record_type import TlsaRecordType
KindType = Literal['cert', 'privkey', 'chain', 'fullchain']
_TargetDictType = dict[KindType, Path]
_LOGGER = getLogger(name=__name__)
class DaneRenewableCert(RenewableCert):
def get_tlsa_bytes(
cert: Certificate, selector: TlsaSelector,
matching_type: TlsaMatchingType
) -> bytes:
'Return a generated `bytes` data of the requested variant.'
if selector == TlsaSelector.ENTIRE:
data = cert.public_bytes(encoding=Encoding.DER)
elif selector == TlsaSelector.PUBLICKEY:
data = cert.public_key().public_bytes(
encoding=Encoding.DER,
format=PublicFormat.SubjectPublicKeyInfo)
if matching_type == TlsaMatchingType.ENTIRE:
return data
elif matching_type == TlsaMatchingType.SHA256:
return sha256(string=data).digest()
elif matching_type == TlsaMatchingType.SHA512:
return sha512(string=data).digest()
class DaneCert(RenewableCert):
"""
Wrapping and extending a `RenewableCert` to have added methods &
functionality.
@ -26,7 +52,8 @@ class DaneRenewableCert(RenewableCert):
_wrapped_renewablecert: RenewableCert
_UPCOMING_DIRNAME = 'upcoming'
__CACHED_PROPERTIES = (
'latest_upcoming_mtime', 'common_names', 'all_upcomings_in_place')
'latest_upcoming_mtime', 'common_names', 'all_upcomings_in_place',
'_DaneCert__upcoming_pubcert', '_DaneCert__pubcert')
def __init__(self):
'Placeholder to override parent\'s `__init__`'
@ -34,25 +61,27 @@ class DaneRenewableCert(RenewableCert):
def __reset_cache(self):
'Resed cached properties.'
for name in self.__CACHED_PROPERTIES:
if hasattr(self, name):
if name in self.__dict__:
delattr(self, name)
self.get_tlsa_recordtypes.cache_clear()
@staticmethod
def from_renewablecert(cert: RenewableCert) -> DaneRenewableCert:
result = DaneRenewableCert()
def from_renewablecert(cert: RenewableCert) -> DaneCert:
'Return a `DaneCert` from a `RenewableCert`.'
result = DaneCert()
result.__dict__ = cert.__dict__
result._wrapped_renewablecert = cert
return result
@staticmethod
def from_data(
config_filename: str, cli_config: NamespaceConfig,
update_symlinks: bool = False
) -> DaneRenewableCert:
def from_configdata(
config_filename: str, cli_config: NamespaceConfig,
update_symlinks: bool = False) -> DaneCert:
'Return a `DaneCert` from configuration data.'
cert = RenewableCert(
config_filename=config_filename, cli_config=cli_config,
update_symlinks=update_symlinks)
return DaneRenewableCert.from_renewablecert(cert=cert)
return DaneCert.from_renewablecert(cert=cert)
@cached_property
def upcoming_dir(self) -> Path:
@ -65,25 +94,26 @@ class DaneRenewableCert(RenewableCert):
self._UPCOMING_DIRNAME).joinpath(original_path.name)
@cached_property
def all_upcomings_in_place(self) -> list[Path]:
def all_upcomings_in_place(self) -> _TargetDictType:
'Return all of the upcoming certificate `Path`s.'
self.upcoming_dir.mkdir(exist_ok=True)
next_version = self.next_free_version()
all_upcoming = [
self.get_upcoming_path(kind=kind, version=next_version)
for kind in ALL_FOUR]
if all(Path.is_file(x) for x in all_upcoming):
all_upcoming: _TargetDictType = {
kind: self.get_upcoming_path(kind=kind, version=next_version)
for kind in ALL_FOUR}
if all(Path.is_file(x) for x in all_upcoming.values()):
return all_upcoming
for upcoming_path in all_upcoming:
for upcoming_path in all_upcoming.values():
upcoming_path.unlink(missing_ok=True)
return []
return {}
@cached_property
def latest_upcoming_mtime(self) -> float:
'Return the latest mtime of the upcoming certificate.'
return max(
(Path.stat(x).st_mtime for x in self.all_upcomings_in_place),
default=0)
return max((
Path.stat(x).st_mtime
for x in self.all_upcomings_in_place.values()
), default=0)
@cached_property
def common_names(self) -> list[str]:
@ -96,7 +126,7 @@ class DaneRenewableCert(RenewableCert):
archive, and link it as the current (latest) certificate.
"""
archive_path = Path(self.archive_dir)
for pem_path in self.all_upcomings_in_place:
for pem_path in self.all_upcomings_in_place.values():
new_path = archive_path.joinpath(pem_path.name)
if pem_path.is_symlink():
# It's a private key symlinked to a former one (pinned)
@ -193,3 +223,36 @@ class DaneRenewableCert(RenewableCert):
targets=targets, new_cert=new_cert, new_chain=new_chain)
self.__update_configfile()
self.__reset_cache()
@cached_property
def __pubcert(self) -> Certificate:
data = Path(self.cert_path).read_bytes()
return load_pem_x509_certificate(data=data)
@cached_property
def __upcoming_pubcert(self) -> Certificate | None:
if not self.all_upcomings_in_place:
return
data = Path(self.all_upcomings_in_place['cert']).read_bytes()
return load_pem_x509_certificate(data=data)
@lru_cache(maxsize=None)
def get_tlsa_recordtypes(
self, usage: TlsaUsage, selector: TlsaSelector,
matching_type: TlsaMatchingType) -> list[TlsaRecordType]:
'Return a list of generated `TlsaRecordType`s.'
result = list[TlsaRecordType]()
data = get_tlsa_bytes(
cert=self.__pubcert, selector=selector,
matching_type=matching_type)
result.append(TlsaRecordType(
usage=usage, selector=selector, matching_type=matching_type,
data=data))
if self.__upcoming_pubcert:
data = get_tlsa_bytes(
cert=self.__upcoming_pubcert, selector=selector,
matching_type=matching_type)
result.append(TlsaRecordType(
usage=usage, selector=selector, matching_type=matching_type,
data=data))
return result

4
src/daneupdate/utils/config_types/configyaml.py
View File

@ -2,9 +2,7 @@
from typing import Literal, TypedDict
from .common import (
LITERAL_TCP_OR_UDP, LITERAL_TSIG_ALGORITHMS, TlsaMatchingType,
TlsaSelector, TlsaUsage)
from .common import LITERAL_TCP_OR_UDP, LITERAL_TSIG_ALGORITHMS
class DefaultsTlsaPortDict(TypedDict):

37
src/daneupdate/utils/crypto.py
View File

@ -1,37 +0,0 @@
from hashlib import sha256, sha512
from cryptography.hazmat.primitives._serialization import (
Encoding, PublicFormat)
from cryptography.x509 import (
BasicConstraints, Certificate, load_pem_x509_certificate)
def sha_digests(content: bytes) -> tuple[str, str]:
'Generate hexadecimal SHA256 and SHA512 hashes for some data.'
hash_sha256 = sha256()
hash_sha256.update(content)
hash_sha512 = sha512()
hash_sha512.update(content)
return hash_sha256.hexdigest(), hash_sha512.hexdigest()
def read_x509_cert(filename: str) -> Certificate:
'Read x509 certificate from file.'
with open(filename, 'rb') as f:
return load_pem_x509_certificate(f.read())
def dane_tlsa_records(cert: Certificate) -> list[str]:
"""
Return list of TLSA record data for the certificate.
Args:
cert: x509 certificate.
"""
bc = cert.extensions.get_extension_for_class(BasicConstraints).value
# DANE-TA=2, DANE-EE=3
usage = 2 if bc.ca else 3
public_key = cert.public_key().public_bytes(
format=PublicFormat.SubjectPublicKeyInfo, encoding=Encoding.DER)
h_sha256, h_sha512 = sha_digests(content=public_key)
return [f'{usage} 1 1 {h_sha256}', f'{usage} 1 2 {h_sha512}']

80
src/daneupdate/utils/tlsa/liveupdate_config.py
View File

@ -3,18 +3,24 @@ from __future__ import annotations
from dataclasses import dataclass
from functools import cached_property
from ipaddress import IPv4Address, IPv6Address
from itertools import product
from certbot.errors import ConfigurationError
from dns.name import Name
from dns.name import from_text as name_from_text
from dns.nameserver import Do53Nameserver
from dns.rdatatype import TLSA
from dns.rdtypes.ANY.TLSA import TLSA as RdTypeTlsa
from dns.resolver import Resolver
from dns.tsig import Key
from dns.tsigkeyring import from_text as tsig_from_text
from dns.update import Update
from ktools.cache.functional import memoized_method
from ..cert import DaneCert
from ..config import (
AdoptedCertConfig, AdoptedCertHostitemRecordsitemConfig,
AdoptedCertHostitemServeritemConfig, Configuration)
from ..config_types.cert import DaneRenewableCert
from ..config_types.common import (
TlsaMatchingType, TlsaProtocol, TlsaSelector, TlsaUsage)
from .record_type import TlsaRecordType
@ -91,12 +97,30 @@ class _DnsServerInfo(object):
result.append(_DnsServerInfo(ip=item.ip, tsig=tsig))
return result
@cached_property
def resolver(self) -> Resolver:
'A cached resolver.'
my_nameserver = Do53Nameserver(address=str(self))
resolver = Resolver(configure=False)
resolver.nameservers = [my_nameserver]
return resolver
def resolve(self, fqdn: str) -> set[TlsaRecordType]:
qname = name_from_text(text=fqdn)
answer = self.resolver.resolve(qname=qname, rdtype=TLSA)
response_set = set[TlsaRecordType]()
item: RdTypeTlsa
for item in answer: # type: ignore
record = TlsaRecordType.from_rdtype(item=item)
response_set.add(record)
return response_set
_serverinfo_cache = dict[Configuration, list[_DnsServerInfo]]()
@dataclass
class _TlsaRecordsInfo(object):
class _TlsaRecordsConfig(object):
'Gathered `runtime.yaml` and `config.yaml` ports.'
protocol: TlsaProtocol
number: int
@ -108,35 +132,52 @@ class _TlsaRecordsInfo(object):
def from_adopted(
records: list[AdoptedCertHostitemRecordsitemConfig],
config: Configuration
) -> list[_TlsaRecordsInfo]:
) -> list[_TlsaRecordsConfig]:
"""
Return prepared records configuration by merging from the
adopted and the default configuration.
"""
result = list[_TlsaRecordsInfo]()
result = list[_TlsaRecordsConfig]()
def_tlsa = config.defaults.tlsa
for item in records:
usage = item.usage or def_tlsa.usage
selector = item.selector or def_tlsa.selector
matching_type = item.matching_type or def_tlsa.matching_type
result.append(_TlsaRecordsInfo(
result.append(_TlsaRecordsConfig(
protocol=item.protocol, number=item.number, usage=usage,
selector=selector, matching_type=matching_type))
return result
@staticmethod
def get_defaults(config: Configuration) -> list[_TlsaRecordsInfo]:
def get_defaults(config: Configuration) -> list[_TlsaRecordsConfig]:
'Return configuration defaults.'
result = list[_TlsaRecordsInfo]()
result = list[_TlsaRecordsConfig]()
def_tlsa = config.defaults.tlsa
for item in def_tlsa.ports:
result.append(_TlsaRecordsInfo(
result.append(_TlsaRecordsConfig(
protocol=TlsaProtocol.from_config_literal(data=item.protocol),
number=item.number, usage=def_tlsa.usage,
selector=def_tlsa.selector,
matching_type=def_tlsa.matching_type))
return result
@cached_property
def prefix(self) -> str:
return f'_{self.number}.{self.protocol.value}'
def required_records_for_cert(self, cert: DaneCert) -> set[TlsaRecordType]:
"""
Return the currently required `TlsaRecordType`s for the passed
`DaneCert`.
"""
result = set[TlsaRecordType]()
iterator = product(self.usage, self.selector, self.matching_type)
for usage, selector, matching_type in iterator:
result.update(cert.get_tlsa_recordtypes(
usage=usage, selector=selector,
matching_type=matching_type))
return result
@dataclass
class HostnameConfig(object):
@ -145,7 +186,7 @@ class HostnameConfig(object):
zone: str
ttl: int
servers: list[_DnsServerInfo]
records: list[_TlsaRecordsInfo]
records: list[_TlsaRecordsConfig]
@staticmethod
def from_fqdn(fqdn: str, config: Configuration) -> HostnameConfig:
@ -158,12 +199,12 @@ class HostnameConfig(object):
return HostnameConfig(
name=host, zone=zone, ttl=config.defaults.tlsa.ttl,
servers=_DnsServerInfo.get_defaults(config=config),
records=_TlsaRecordsInfo.get_defaults(config=config))
records=_TlsaRecordsConfig.get_defaults(config=config))
@staticmethod
def from_adoptedhosts(
adopted: AdoptedCertConfig, config: Configuration,
cert: DaneRenewableCert
cert: DaneCert
) -> list[HostnameConfig]:
"""
Return a list of `LiveupdateConfig` from adopted runtime
@ -180,14 +221,23 @@ class HostnameConfig(object):
servers = _DnsServerInfo.from_adopted(
servers=item.servers, config=config)
if item.records is None:
records = _TlsaRecordsInfo.get_defaults(config=config)
records = _TlsaRecordsConfig.get_defaults(config=config)
else:
records = _TlsaRecordsInfo.from_adopted(
records = _TlsaRecordsConfig.from_adopted(
records=item.records, config=config)
result.append(HostnameConfig(
name=item.name, zone=zone, ttl=ttl, servers=servers,
records=records))
return result
def tlsa_records(self) -> list[TlsaRecordType]:
pass
def get_tlsa_records(
self, cert: DaneCert) -> dict[str, set[TlsaRecordType]]:
'Return TLSA records for the configuration here.'
result = dict[str, set[TlsaRecordType]]()
for record in self.records:
name = record.prefix
if self.name != '.':
name += f'.{self.name}'
name += f'.{self.zone}'
result[name] = record.required_records_for_cert(cert=cert)
return result

8
src/daneupdate/utils/tlsa/record_type.py
View File

@ -16,22 +16,22 @@ class TlsaRecordType(object):
usage: TlsaUsage
selector: TlsaSelector
matching_type: TlsaMatchingType
cert: bytes
data: bytes
@staticmethod
def from_rdtype(item: TLSA):
return TlsaRecordType(
usage=TlsaUsage(value=item.usage),
selector=TlsaSelector(value=item.selector),
matching_type=TlsaMatchingType(value=item.mtype), cert=item.cert)
matching_type=TlsaMatchingType(value=item.mtype), data=item.cert)
def __eq__(self, other: object) -> bool:
if isinstance(other, TlsaRecordType):
return self.usage == other.usage \
and self.selector == other.selector \
and self.matching_type == other.matching_type \
and self.cert == other.cert
and self.data == other.data
return False
def __hash__(self) -> int:
return hash((self.usage, self.selector, self.matching_type, self.cert))
return hash((self.usage, self.selector, self.matching_type, self.data))

38
src/daneupdate/utils/tlsa/updater.py
View File

@ -1,14 +1,31 @@
from daneupdate.utils.config_types.cert import DaneRenewableCert
from daneupdate.utils.cert import DaneCert
from daneupdate.utils.tlsa.record_type import TlsaRecordType
from dns.name import from_text as name_from_text
from dns.rdatatype import TLSA
from dns.rdtypes.ANY.TLSA import TLSA as RdTypeTlsa
from dns.resolver import Resolver
from ..config import Configuration
from .liveupdate_config import HostnameConfig
def _get_records_from_ns(
fqdn: str, resolver: Resolver) -> set[TlsaRecordType]:
qname = name_from_text(text='_25._tcp.mail.weberdns.de.')
answer = resolver.resolve(qname=qname, rdtype=TLSA)
response_set = set[TlsaRecordType]()
item: RdTypeTlsa
for item in answer:
record = TlsaRecordType.from_rdtype(item=item)
response_set.add(record)
return response_set
class TlsaUpdater(object):
'Updating TLSA records for an deployed and/or upcoming certificate.'
def __init__(
self, config: Configuration, cert: DaneRenewableCert,
self, config: Configuration, cert: DaneCert,
certconfig_path: str
):
self._adopted_config = config.runtime.adopted[certconfig_path]
@ -28,13 +45,16 @@ class TlsaUpdater(object):
HostnameConfig.from_fqdn(fqdn=fqdn, config=self._config)
for fqdn in self._cert.names()]
def _get_missing_records(self):
pass
def process(self):
if self._config.defaults.method == 'dns-update':
for hostname_config in self._get_hostname_configs():
for serverinfo in hostname_config.servers:
update = serverinfo.get_zoneupdate(
zone=hostname_config.zone)
update.replace
tlsa_records = \
hostname_config.get_tlsa_records(cert=self._cert)
for fqdn, records_required in tlsa_records.items():
for server in hostname_config.servers:
records_existing = server.resolve(fqdn=fqdn)
if records_existing != records_required:
pass
# update = serverinfo.get_zoneupdate(
# zone=hostname_config.zone)
# update.replace