Fix subdomain filtering

2 files changed, 31 insertions, 3 deletions

diff --git a/src/scadere/listen.py b/src/scadere/listen.py
index 982f547..6dc8f3a 100644
--- a/src/scadere/listen.py
+++ b/src/scadere/listen.py
@@ -127,6 +127,20 @@ async def write_xml(writer, document):
         await writer.drain()
 
 
+def split_domain(domain):
+    """Split domain and order by ascending level."""
+    return tuple(domain.split('.')[::-1])
+
+
+def is_subdomain(subject, objects):
+    """Check if subject is a subdomain of any object."""
+    if not objects:
+        return True
+    sbj_parts = split_domain(subject)
+    return any(sbj_parts[:len(obj_parts)] == obj_parts
+               for obj_parts in map(split_domain, objects))
+
+
 async def handle(certs, base_url, reader, writer):
     """Handle HTTP request."""
     summaries = map(parse_summary, certs.read_text().splitlines())
@@ -138,7 +152,7 @@ async def handle(certs, base_url, reader, writer):
     request = await reader.readuntil(b'\r\n')
     url = request.removeprefix(b'GET ').rsplit(b' HTTP/', 1)[0].strip()
     url_parts = urlsplit(urljoin(base_url, url.decode()))
-    domains = tuple(parse_qs(url_parts.query).get('domain', ['']))
+    domains = tuple(parse_qs(url_parts.query).get('domain', []))
 
     if not request.startswith(b'GET '):
         await describe_status(writer, HTTPStatus.METHOD_NOT_ALLOWED)
@@ -155,7 +169,7 @@ async def handle(certs, base_url, reader, writer):
                   'version': __version__},
                  'Scadere'),
                 *(entry(base_url, cert) for cert in lookup.values()
-                  if cert[2].endswith(domains)))
+                  if is_subdomain(cert[2], domains)))
         await write_xml(writer, feed)
     elif url_parts.path in lookup:  # accessible Atom entry's link/ID
         await write_status(writer, HTTPStatus.OK)
diff --git a/tst/test_listen.py b/tst/test_listen.py
index c6d9cd4..3737baa 100644
--- a/tst/test_listen.py
+++ b/tst/test_listen.py
@@ -36,7 +36,8 @@ from hypothesis.strategies import (builds, composite, data,
                                    datetimes, integers, lists, text)
 from hypothesis.provisional import domains, urls
 
-from scadere.listen import body, entry, handle, path, with_trailing_slash, xml
+from scadere.listen import (body, entry, handle, is_subdomain,
+                            path, with_trailing_slash, xml)
 
 ATOM_NAMESPACES = {'': 'http://www.w3.org/2005/Atom'}
 XHTML_NAMESPACES = {'': 'http://www.w3.org/1999/xhtml'}
@@ -119,6 +120,19 @@ def test_atom_entry(base_url, hostname, port,
 </entry>'''
 
 
+@given(domains(), lists(domains()))
+def test_is_subdomain(subject, objects):
+    if not objects:
+        assert is_subdomain(subject, objects)
+    elif is_subdomain(subject, objects):
+        assert any(child == '' or child.endswith('.')
+                   for child in map(subject.removesuffix, objects))
+    else:
+        for obj in objects:
+            assert (not subject.endswith(obj)
+                    or not subject.removesuffix(obj).endswith('.'))
+
+
 @composite
 def certificates(draw):
     """Return a Hypothesis strategy for certificate summaries."""