webfinger/__init__.py

@@ -2,31 +2,32 @@ from fastapi import FastAPI
 import webfinger.lookup
 
 app = FastAPI(
-	title = "webfinger"
+	title = "webfinger",
+	version = "2"
 )
+
 app.include_router(webfinger.lookup.router)
 
-from fastapi.openapi.utils import generate_operation_id
-from fastapi.routing import APIRoute
 
-def remove_422s(app: FastAPI) -> None:
+def cleanup_openapi(app: FastAPI) -> None:
+	"""Cleanup inconsistencies in the OpenAPI documentation"""
 	openapi_schema = app.openapi()
-    operation_ids_to_update: set[str] = set()
-    for route in app.routes:
-        if not isinstance(route, APIRoute):
-            continue
-        methods = route.methods or ["GET"]
-        if getattr(route.endpoint, "__remove_422__", None):
-            for method in methods:
-                operation_ids_to_update.add(generate_operation_id(route=route, method=method))
-    paths = openapi_schema["paths"]
-    for _, operations in paths.items():
-        for method, metadata in operations.items():
-            operation_id = metadata.get("operationId")
-            if operation_id in operation_ids_to_update:
-                metadata["responses"].pop("422", None)
 
-remove_422s(app)
+	# Remove 422 Validation Error from Webfinger lookup (RFC7033 uses 400 Bad Request)
+	openapi_schema["paths"]["/.well-known/webfinger"]["get"]["responses"].pop("422")
+
+	# Remove 422 Validation Error from schemas
+	schemas = openapi_schema["components"]["schemas"]
+	schemas.pop("ValidationError")
+	schemas.pop("HTTPValidationError")
+
+	# Mark resource param as required
+	lookup_params = openapi_schema["paths"]["/.well-known/webfinger"]["get"]["parameters"]
+	resource_param = [param for param in lookup_params if param["name"] == "resource"][0]
+	resource_param["required"] = True
+
+
+cleanup_openapi(app)
 
 if __name__ == "__main__":
 	import uvicorn

webfinger/io.py

@@ -0,0 +1,18 @@
+import json
+from os import environ as env
+from pathlib import Path
+
+def get_document(resource: str) -> dict | None:
+	# 
+	dir = env.get("RESOURCE_DIR") or "resource"
+	filename = f"{dir}/{resource}.json"
+	path = Path(filename)
+
+	if not path:
+		return None
+
+	# open the local file
+	with open(filename, "r") as file:
+		data: dict = json.loads(file.read())
+
+	return data

webfinger/lookup.py

@@ -1,11 +1,11 @@
 from fastapi import APIRouter, Response
 from fastapi.responses import PlainTextResponse, JSONResponse
 from fastapi.params import Query
-import json
+
-from pathlib import Path
-from os import environ as env
 from pydantic import BaseModel
 
+from webfinger.io import get_document
+
 
 router = APIRouter(
 	prefix="/.well-known/webfinger"
@@ -34,6 +34,8 @@ EXAMPLE_RESOURCE = r'''
 }
 '''
 
+NO_RESOURCE_HINT = "Query ?resource={resource} to obtain information about that resource."
+
 
 ## Pydantic models for OpenAPI schema
 
@@ -74,7 +76,7 @@ RESPONSES = {
 		"description": "Bad Request. No resource was provided.",
 		"content": {
 			"text/plain": {
-				"example": "Query ?resource= for more information."
+				"example": NO_RESOURCE_HINT
 			}
 		},
 	},
@@ -100,10 +102,6 @@ RESPONSES = {
 
 ## The lookup method as described in RFC7033
 
-def remove_422(func):
-	func.__remove_422__ = True
-	return func
-
 @router.get(
 	"",
 	summary = "Lookup a Webfinger resource",
@@ -114,7 +112,6 @@ def remove_422(func):
 	responses = {**RESPONSES},
 
 )
-@remove_422
 async def lookup(
 	resource: str = Query(
 		None,
@@ -129,40 +126,37 @@ async def lookup(
 	Respond to a WebFinger query.
 	"""
 
-	# Basic info as a hint, if no resource given
+	# If no resource is given, then show a basic hint.
 	if not resource:
 		return PlainTextResponse(
-			content="Query ?resource= for more information.",
+			content=NO_RESOURCE_HINT,
 			status_code=400,
 		)
 	
-	# otherwise, load that resource and return its data
+	# Otherwise, try to read the resource document.
-	dir = env.get("RESOURCE_DIR") or "resource"
+	data = get_document(resource)
-	filename = f"{dir}/{resource}.json"
-	path = Path(filename)
 
-	# TODO redirect to upstream webfinger server(s) if resource not found locally
+	# If the document could not be read, then it likely didn't exist.
-	#       and if there is an upstream to try
+	if data is None:
-	if not path.is_file():
 		return PlainTextResponse(
 			content=f"{resource} not found",
 			status_code=404,
 		)
 
-	# open the local file
+	# Obtain values from the resource document.
-	with open(filename, "r") as file:
+	aliases = [alias for alias in data.get('aliases', [])]
-		data: dict = json.loads(file.read())
-
-	# filter links if rel is provided
 	links = [link for link in data.get('links', [])]
+	properties = data.get('properties', {})
+
+	# If optional rel is provided, then filter only for links with that rel-type.
 	if rel:
 		links = [link for link in links if link.get('rel', '') in rel]
 
-	# construct json response
+	# Construct JSON response and return it.
 	content = {
 		"subject": resource,
-		"aliases": [alias for alias in data.get('aliases')],
+		"aliases": aliases,
-		"properties": data.get('properties'),
+		"properties": properties,
 		"links": links,
 	}
 	content = {k: v for k, v in content.items() if v} # remove null values