From 02d708b8e35fa756b6ea14b4fc132341783ebd5e Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Robin=20Hru=C5=A1ka?= <byewokko@seznam.cz>
Date: Thu, 29 Feb 2024 17:09:06 +0100
Subject: [PATCH] handle connection errors in FIDO MDS initialization; add
 switch for disabling MDS

---
 seacatauth/authn/webauthn/service.py | 45 +++++++++++++++++++++-------
 1 file changed, 34 insertions(+), 11 deletions(-)

diff --git a/seacatauth/authn/webauthn/service.py b/seacatauth/authn/webauthn/service.py
index 062315c65..ff8af4be5 100644
--- a/seacatauth/authn/webauthn/service.py
+++ b/seacatauth/authn/webauthn/service.py
@@ -28,7 +28,14 @@
 
 asab.Config.add_defaults({
 	"seacatauth:webauthn": {
-		"attestation": "direct"
+		"attestation": "direct",
+
+		# Authenticator metadata source file, for details see https://fidoalliance.org/metadata
+		# Possible values:
+		#   - HTTPS or HTTP address of a JWT MDS file (defaults to "https://mds3.fidoalliance.org")
+		#   - Relative or absolute path of a JWT MDS file
+		#   - Empty value or "DISABLED" (disables the metadata service)
+		"metadata_service_url": "https://mds3.fidoalliance.org",
 	}
 })
 
@@ -37,7 +44,6 @@ class WebAuthnService(asab.Service):
 	WebAuthnCredentialCollection = "wa"
 	WebAuthnRegistrationChallengeCollection = "warc"
 	FidoMetadataServiceCollection = "fms"
-	FidoMetadataServiceUrl = "https://mds3.fidoalliance.org"
 
 	def __init__(self, app, service_name="seacatauth.WebAuthnService"):
 		super().__init__(app, service_name)
@@ -69,31 +75,43 @@ def __init__(self, app, service_name="seacatauth.WebAuthnService"):
 			webauthn.helpers.structs.COSEAlgorithmIdentifier(-7)  # Es256
 		]
 
+		self.FidoMetadataServiceUrl = asab.Config.get("seacatauth:webauthn", "metadata_service_url")
+		if self.FidoMetadataServiceUrl in ("", "DISABLED"):
+			self.FidoMetadataServiceUrl = None
 		self._FidoMetadataByAAGUID: typing.Dict[int, dict] | None = None
 
 		app.PubSub.subscribe("Application.housekeeping!", self._on_housekeeping)
 
 
 	async def initialize(self, app):
-		await self._load_fido_metadata()
+		try:
+			await self._load_fido_metadata()
+		except Exception as e:
+			L.exception("Failed to fetch FIDO Alliance Metadata ({}: {})".format(e.__class__.__name__, e))
 
 
 	async def _on_housekeeping(self, event_name):
 		await self._delete_expired_challenges()
-		await self._load_fido_metadata(only_if_empty=False)
+		try:
+			await self._load_fido_metadata(force_reload=True)
+		except Exception as e:
+			L.info("Failed to fetch FIDO Alliance Metadata: {}".format(e))
 
 
-	async def _load_fido_metadata(self, only_if_empty=True):
+	async def _load_fido_metadata(self, *, force_reload: bool = False):
 		"""
 		Download and decode FIDO metadata from FIDO Alliance Metadata Service (MDS) and prepare a lookup dictionary.
 		"""
-		if only_if_empty:
+		if not self.FidoMetadataServiceUrl:
+			return
+
+		if not force_reload:
 			coll = await self.StorageService.collection(self.FidoMetadataServiceCollection)
 			count = await coll.estimated_document_count()
 			if count > 0:
 				return
 
-		try:
+		if self.FidoMetadataServiceUrl.startswith("https://") or self.FidoMetadataServiceUrl.startswith("http://"):
 			async with aiohttp.ClientSession() as session:
 				async with session.get(self.FidoMetadataServiceUrl) as resp:
 					if resp.status != 200:
@@ -101,9 +119,10 @@ async def _load_fido_metadata(self, only_if_empty=True):
 						L.error("Failed to fetch FIDO metadata:\n{}".format(text[:1000]))
 						return
 					jwt = await resp.text()
-		except ConnectionError:
-			L.error("Cannot connect to FIDO Alliance Metadata Service.")
-			return
+		else:
+			# Load from local file
+			with open(self.FidoMetadataServiceUrl) as f:
+				jwt = f.read()
 
 		jwt = jwcrypto.jwt.JWT(jwt=jwt)
 		cert_chain = jwt.token.jose_header.get("x5c", [])
@@ -179,7 +198,11 @@ async def create_webauthn_credential(
 		L.log(asab.LOG_NOTICE, "WebAuthn credential created", struct_data={"wacid": wacid.hex()})
 
 	async def _get_authenticator_metadata(self, verified_registration):
-		await self._load_fido_metadata()
+		try:
+			await self._load_fido_metadata()
+		except Exception as e:
+			L.info("Failed to fetch FIDO Alliance Metadata: {}".format(e))
+
 		aaguid = bytes.fromhex(verified_registration.aaguid.replace("-", ""))
 		if aaguid == 0:
 			# Authenticators with other identifiers than AAGUID are not supported