work in progress: feat(api,webapp): adds TLS identity manager for aut…

…omatically generating TLSA records

api/desecapi/migrations/0015_identities.py

@@ -0,0 +1,29 @@
+# Generated by Django 3.1.5 on 2021-01-30 15:24
+
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+import uuid
+
+
+class Migration(migrations.Migration):
+
+ dependencies = [
+ ('desecapi', '0014_replication'),
+ ]
+
+ operations = [
+ migrations.CreateModel(
+ name='TLSIdentity',
+ fields=[
+ ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
+ ('name', models.CharField(max_length=24)),
+ ('created', models.DateTimeField(auto_now_add=True)),
+ ('certificate', models.TextField()),
+ ('owner', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='identities', to=settings.AUTH_USER_MODEL)),
+ ],
+ options={
+ 'abstract': False,
+ },
+ ),
+ ]

api/desecapi/migrations/0016_identities.py

@@ -0,0 +1,43 @@
+# Generated by Django 3.1.5 on 2021-01-31 13:00
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+ dependencies = [
+ ('desecapi', '0015_identities'),
+ ]
+
+ operations = [
+ migrations.AddField(
+ model_name='tlsidentity',
+ name='port',
+ field=models.IntegerField(default=443),
+ ),
+ migrations.AddField(
+ model_name='tlsidentity',
+ name='protocol',
+ field=models.TextField(choices=[('tcp', 'Tcp'), ('udp', 'Udp'), ('sctp', 'Sctp')], default='tcp'),
+ ),
+ migrations.AddField(
+ model_name='tlsidentity',
+ name='scheduled_removal',
+ field=models.DateTimeField(null=True),
+ ),
+ migrations.AddField(
+ model_name='tlsidentity',
+ name='tlsa_certificate_usage',
+ field=models.IntegerField(choices=[(0, 'Ca Constraint'), (1, 'Service Certificate Constraint'), (2, 'Trust Anchor Assertion'), (3, 'Domain Issued Certificate')], default=3),
+ ),
+ migrations.AddField(
+ model_name='tlsidentity',
+ name='tlsa_matching_type',
+ field=models.IntegerField(choices=[(0, 'No Hash Used'), (1, 'Sha256'), (2, 'Sha512')], default=1),
+ ),
+ migrations.AddField(
+ model_name='tlsidentity',
+ name='tlsa_selector',
+ field=models.IntegerField(choices=[(0, 'Full Certificate'), (1, 'Subject Public Key Info')], default=1),
+ ),
+ ]

api/desecapi/migrations/0017_auto_20210213_1840.py

@@ -0,0 +1,33 @@
+# Generated by Django 3.1.6 on 2021-02-13 18:40
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+ dependencies = [
+ ('desecapi', '0016_identities'),
+ ]
+
+ operations = [
+ migrations.AddField(
+ model_name='tlsidentity',
+ name='default_ttl',
+ field=models.PositiveIntegerField(default=300),
+ ),
+ migrations.AlterField(
+ model_name='tlsidentity',
+ name='tlsa_certificate_usage',
+ field=models.IntegerField(choices=[(0, 'Ca Constraint'), (1, 'Service Certificate Constraint'), (2, 'Trust Anchor Assertion'), (3, 'Domain Issued Certificate')], default=3),
+ ),
+ migrations.AlterField(
+ model_name='tlsidentity',
+ name='tlsa_matching_type',
+ field=models.IntegerField(choices=[(0, 'No Hash Used'), (1, 'Sha256'), (2, 'Sha512')], default=1),
+ ),
+ migrations.AlterField(
+ model_name='tlsidentity',
+ name='tlsa_selector',
+ field=models.IntegerField(choices=[(0, 'Full Certificate'), (1, 'Subject Public Key Info')], default=1),
+ ),
+ ]

api/desecapi/models.py

@@ -12,10 +12,12 @@
 from datetime import timedelta
 from functools import cached_property
 from hashlib import sha256
+from typing import Set, List, Optional, Tuple, Dict
 
 import dns
 import psl_dns
 import rest_framework.authtoken.models
+from cryptography import x509, hazmat
 from django.conf import settings
 from django.contrib.auth.hashers import make_password
 from django.contrib.auth.models import AbstractBaseUser, AnonymousUser, BaseUserManager
@@ -946,3 +948,209 @@ def verify(self, solution: str):
  and
  age <= settings.CAPTCHA_VALIDITY_PERIOD # not expired
  )
+
+
+class Identity(models.Model):
+ rr_type = None
+
+ id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
+ name = models.CharField(max_length=24, default="")
+ created = models.DateTimeField(auto_now_add=True)
+ owner = models.ForeignKey(User, on_delete=models.PROTECT, related_name='identities')
+ default_ttl = models.PositiveIntegerField(default=300)
+
+ class Meta:
+ abstract = True
+
+ def get_record_contents(self) -> List[str]:
+ raise NotImplementedError
+
+ def save_rrs(self):
+ raise NotImplementedError
+
+ def save(self, *args, **kwargs):
+ self.save_rrs()
+ return super().save(*args, **kwargs)
+
+ def delete_rrs(self):
+ raise NotImplementedError
+
+ def delete(self, using=None, keep_parents=False):
+ # TODO this will delete also RRs that may be covered by other identities
+ self.delete_rrs()
+ return super().delete(using, keep_parents)
+
+ def get_or_create_rr_set(self, domain: Domain, subname: str) -> RRset:
+ try:
+ return RRset.objects.get(domain=domain, subname=subname, type=self.rr_type)
+ except RRset.DoesNotExist:
+ # TODO save this RRset?
+ return RRset(domain=domain, subname=subname, type=self.rr_type, ttl=self.default_ttl)
+
+ @staticmethod
+ def get_or_create_rr(rrset: RRset, content: str) -> RR:
+ try:
+ return RR.objects.get(rrset=rrset, content=content)
+ except RR.DoesNotExist:
+ return RR(rrset=rrset, content=content)
+
+
+class TLSIdentity(Identity):
+ rr_type = 'TLSA'
+
+ class CertificateUsage(models.IntegerChoices):
+ CA_CONSTRAINT = 0
+ SERVICE_CERTIFICATE_CONSTRAINT = 1
+ TRUST_ANCHOR_ASSERTION = 2
+ DOMAIN_ISSUED_CERTIFICATE = 3
+
+ class Selector(models.IntegerChoices):
+ FULL_CERTIFICATE = 0
+ SUBJECT_PUBLIC_KEY_INFO = 1
+
+ class MatchingType(models.IntegerChoices):
+ NO_HASH_USED = 0
+ SHA256 = 1
+ SHA512 = 2
+
+ class Protocol(models.TextChoices):
+ TCP = 'tcp'
+ UDP = 'udp'
+ SCTP = 'sctp'
+
+ certificate = models.TextField()
+
+ tlsa_selector = models.IntegerField(choices=Selector.choices, default=Selector.SUBJECT_PUBLIC_KEY_INFO)
+ tlsa_matching_type = models.IntegerField(choices=MatchingType.choices, default=MatchingType.SHA256)
+ tlsa_certificate_usage = models.IntegerField(choices=CertificateUsage.choices,
+ default=CertificateUsage.DOMAIN_ISSUED_CERTIFICATE)
+
+ port = models.IntegerField(default=443)
+ protocol = models.TextField(choices=Protocol.choices, default=Protocol.TCP)
+
+ scheduled_removal = models.DateTimeField(null=True)
+
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+ if 'not_valid_after' not in kwargs:
+ self.scheduled_removal = self.not_valid_after
+
+ def get_record_contents(self) -> List[str]:
+ # choose hash function
+ if self.tlsa_matching_type == self.MatchingType.SHA256:
+ hash_function = hazmat.primitives.hashes.SHA256()
+ elif self.tlsa_matching_type == self.MatchingType.SHA512:
+ hash_function = hazmat.primitives.hashes.SHA512()
+ else:
+ raise NotImplementedError
+
+ # choose data to hash
+ if self.tlsa_selector == self.Selector.SUBJECT_PUBLIC_KEY_INFO:
+ to_be_hashed = self._cert.public_key().public_bytes(
+ hazmat.primitives.serialization.Encoding.DER,
+ hazmat.primitives.serialization.PublicFormat.SubjectPublicKeyInfo
+ )
+ else:
+ raise NotImplementedError
+
+ # compute the hash
+ h = hazmat.primitives.hashes.Hash(hash_function)
+ h.update(to_be_hashed)
+ hash = h.finalize().hex()
+
+ # create TLSA record content
+ return [f"{self.tlsa_certificate_usage} {self.tlsa_selector} {self.tlsa_matching_type} {hash}"]
+
+ @property
+ def _cert(self) -> x509.Certificate:
+ return x509.load_pem_x509_certificate(self.certificate.encode())
+
+ @property
+ def fingerprint(self) -> str:
+ return self._cert.fingerprint(hazmat.primitives.hashes.SHA256()).hex()
+
+ @property
+ def subject_names(self) -> Set[str]:
+ subject_names = {
+ x.value for x in
+ self._cert.subject.get_attributes_for_oid(x509.oid.NameOID.COMMON_NAME)
+ }
+
+ try:
+ subject_alternative_names = {
+ x for x in
+ self._cert.extensions.get_extension_for_oid(
+ x509.oid.ExtensionOID.SUBJECT_ALTERNATIVE_NAME).value.get_values_for_type(x509.DNSName)
+ }
+ except x509.extensions.ExtensionNotFound:
+ subject_alternative_names = set()
+
+ return subject_names | subject_alternative_names
+
+ @staticmethod
+ def get_closest_ancestor(domain_name, owner: User) -> Optional[Domain]:
+ # TODO move to Domain?
+ labels = domain_name.split('.')
+ ancestor_names = ['.'.join(labels[i:]) for i in range(len(labels))]
+ for ancestor_name in ancestor_names: # TODO do this with one query
+ try:
+ return Domain.objects.get(name=ancestor_name, owner=owner)
+ except Domain.DoesNotExist:
+ continue
+ return None
+
+ def domains_subnames(self) -> Set[Tuple[Domain, str]]:
+ domains_subnames = set()
+ for name in self.subject_names:
+ # cut off any wildcard prefix
+ name = name.lstrip('*').lstrip('.')
+
+ # filter names for valid domain names
+ try:
+ validate_domain_name[1](name)
+ except ValidationError:
+ continue
+
+ # find user-owned parent domain
+ domain = self.get_closest_ancestor(name, self.owner)
+ if not domain:
+ continue
+ subname = name[:-len(domain.name)].rstrip('.')
+
+ # return subname, domain pair
+ domains_subnames.add((domain, f"_{self.port:n}._{self.protocol}.{subname}".rstrip('.')))
+ return domains_subnames
+
+ def get_rrsets(self) -> List[RRset]:
+ rrsets = []
+ for domain, subname in self.domains_subnames():
+ rrsets.append(self.get_or_create_rr_set(domain, subname))
+ return rrsets
+
+ def get_rrs(self) -> List[RR]:
+ rrs = []
+ for domain, subname in self.domains_subnames():
+ rrset = self.get_or_create_rr_set(domain, subname)
+ for content in self.get_record_contents():
+ rrs.append(self.get_or_create_rr(rrset=rrset, content=content))
+ return rrs
+
+ def save_rrs(self):
+ for rr in self.get_rrs():
+ rr.rrset.save()
+ rr.save()
+
+ def delete_rrs(self):
+ for domain, subname in self.domains_subnames():
+ rrset = self.get_or_create_rr_set(domain, subname)
+ rrset.records.filter(content__in=self.get_record_contents()).delete()
+ if not len(rrset.records.all()):
+ rrset.delete()
+
+ @property
+ def not_valid_before(self):
+ return self._cert.not_valid_before
+
+ @property
+ def not_valid_after(self):
+ return self._cert.not_valid_after

api/desecapi/serializers.py

@@ -849,3 +849,31 @@ class AuthenticatedRenewDomainBasicUserActionSerializer(AuthenticatedDomainBasic
 
  class Meta(AuthenticatedDomainBasicUserActionSerializer.Meta):
  model = models.AuthenticatedRenewDomainBasicUserAction
+
+
+class TLSIdentitySerializer(serializers.ModelSerializer):
+ published_at = serializers.SerializerMethodField(read_only=True)
+
+ def get_published_at(self, tls_identity: models.TLSIdentity):
+ return [
+ f"{rrset.type}/{rrset.name}"
+ for rrset in tls_identity.get_rrsets()
+ ]
+
+ class Meta:
+ model = models.TLSIdentity
+ fields = (
+ 'id', 'name', 'created',
+
+ 'default_ttl',
+
+ 'certificate',
+ 'tlsa_selector', 'tlsa_matching_type', 'tlsa_certificate_usage',
+
+ 'port', 'protocol',
+
+ 'fingerprint', 'not_valid_before', 'not_valid_after', 'subject_names',
+
+ 'published_at',
+ )
+ read_only_fields = list(filter(lambda f: f not in ('name', 'certificate'), fields))