Split the various methods we check caldav urls into separate functions

backend/src/appointment/controller/calendar.py

@@ -290,7 +290,7 @@ def __init__(self, db: Session, subscriber_id: int, calendar_id: int, redis_inst
 
  self.db = db
  self.provider = CalendarProvider.caldav
- self.url = Tools.fix_caldav_urls(url)
+ self.url = url
  self.password = password
  self.user = user
 
@@ -732,31 +732,53 @@ def existing_events_for_schedule(
  return existing_events
 
  @staticmethod
- def dns_caldav_lookup(url):
+ def dns_caldav_lookup(url, secure=True):
  import dns.resolver
 
+ secure_character = 's' if secure else ''
+ dns_url = f'_caldav{secure_character}._tcp.{url}'
+ path = ''
+
  # Check if they have a caldav subdomain, on error just return none
  try:
- records = dns.resolver.resolve(f'_caldavs._tcp.{url}', 'SRV')
+ records = dns.resolver.resolve(dns_url, 'SRV')
  except DNSException:
  return None, None
 
+ # Check if they have any relative paths setup
+ try:
+ txt_records = dns.resolver.resolve(dns_url, 'TXT')
+
+ for txt_record in txt_records:
+ # Remove any quotes from the txt record
+ txt_record = str(txt_record).replace('"', '')
+ if txt_record.startswith('path='):
+ path = txt_record[5:]
+ except DNSException:
+ pass
+
+ # Append a slash at the end if it's missing
+ path += '' if path.endswith('/') else '/'
+
  # Grab the first item or None
  caldav_host = None
+ port = 443 if secure else 80
  ttl = records.rrset.ttl or 300
  if len(records) > 0:
+ port = str(records[0].port)
  caldav_host = str(records[0].target)[:-1]
 
+ if '://' in caldav_host:
+ # Remove any protocols first!
+ caldav_host.replace('http://', '').replace('https://', '')
+
  # We should only be pulling the secure link
- if '://' not in caldav_host:
- caldav_host = f'https://{caldav_host}'
+ caldav_host = f'http{secure_character}://{caldav_host}:{port}{path}'
 
  return caldav_host, ttl
 
  @staticmethod
- def fix_caldav_urls(url: str) -> str:
- """Fix up some common url issues with some caldav providers"""
-
+ def well_known_caldav_lookup(url: str) -> str|None:
  parsed_url = urlparse(url)
 
  # Do they have a well-known?
@@ -771,6 +793,12 @@ def fix_caldav_urls(url: str) -> str:
  except requests.exceptions.ConnectionError:
  # Ignore connection errors here
  pass
+ return None
+
+ @staticmethod
+ def fix_caldav_urls(url: str) -> str:
+ """Fix up some common url issues with some caldav providers"""
+ parsed_url = urlparse(url)
 
  # Handle any fastmail issues
  if 'fastmail.com' in parsed_url.hostname:

backend/src/appointment/routes/caldav.py

@@ -43,13 +43,23 @@ def caldav_autodiscover_auth(
  if lookup_url is None:
  parsed_url = urlparse(connection.url)
  lookup_url, ttl = Tools.dns_caldav_lookup(parsed_url.hostname)
- if lookup_url:
- connection.url = lookup_url
- # set the cached lookup for the remainder of the dns ttl
- if redis_client:
- redis_client.set(dns_lookup_cache_key, utils.encrypt(lookup_url), ex=ttl)
+ # set the cached lookup for the remainder of the dns ttl
+ if redis_client and lookup_url:
+ redis_client.set(dns_lookup_cache_key, utils.encrypt(lookup_url), ex=ttl)
  else:
- connection.url = utils.decrypt(lookup_url)
+ # Extract the cached value
+ lookup_url = utils.decrypt(lookup_url)
+
+ # Check for well-known
+ if lookup_url is None:
+ lookup_url = Tools.well_known_caldav_lookup(connection.url)
+
+ # If we have a lookup_url then apply it
+ if lookup_url:
+ connection.url = lookup_url
+
+ # Finally perform any final fixups needed
+ connection.url = Tools.fix_caldav_urls(connection.url)
 
  con = CalDavConnector(
  db=db,

backend/test/unit/test_calendar_tools.py

@@ -70,3 +70,21 @@ def test_meeting_url_in_location(self, with_db, make_google_calendar, make_appoi
  ics = Tools().create_vevent(appointment, slot, subscriber)
  assert ics
  assert ':'.join(['LOCATION', slot.meeting_link_url])
+
+
+class TestDnsCaldavLookup:
+ def test_for_host(self):
+ host, ttl = Tools.dns_caldav_lookup('thunderbird.net')
+ assert host == 'https://apidata.googleusercontent.com:443/'
+ assert ttl
+
+ def test_for_txt_record(self):
+ """This domain is used with permission from the owner (me, melissa autumn!)"""
+ host, ttl = Tools.dns_caldav_lookup('melissaautumn.com')
+ assert host == 'https://caldav.fastmail.com:443/dav/'
+ assert ttl
+
+ def test_no_records(self):
+ host, ttl = Tools.dns_caldav_lookup('appointment.day')
+ assert host is None
+ assert ttl is None