webfinger/webfinger/lookup.py

from fastapi import APIRouter, Response
from fastapi.responses import PlainTextResponse, JSONResponse
from fastapi.params import Query

from pydantic import BaseModel
import json

from webfinger.io import get_jrd


router = APIRouter(
	prefix="/.well-known/webfinger",
	)


# Define required CORS header per RFC7033

HEADERS = {'Access-Control-Allow-Origin': '*'}
# if env.get("CORS_ALLOW_ORIGIN"):
# 	allowed_origins = env.get("CORS_ALLOW_ORIGIN")
# 	headers.update({'Access-Control-Allow-Origin': allowed_origins})


## Pydantic models for OpenAPI schema

class Link(BaseModel):
	rel: str
	type: str | None
	href: str | None
	titles: str | None
	properties: str | None

	class Config:
		schema_extra = {
			"example": r'''{
				"rel": "https://webfinger.net/rel/profile-page/",
				"type": "text/html",
				"href": "https://trwnh.com/~a"
			}'''
		}

class JRD(BaseModel):
	subject: str
	aliases: list[str] | None
	links: list[Link] | None

	class Config:
		schema_extra = {
			"example": r'''{
				"subject": "acct:a@trwnh.com",
				"aliases": ["https://trwnh.com/actors/7057bc10-db1c-4ebe-9e00-22cf04be4e5e", "https://trwnh.com/~a", "acct:trwnh@trwnh.com"],
				"links": [
					{
						"rel": "self",
						"type": "application/ld+json; profile=\"https://www.w3.org/ns/activitystreams\"",
						"href": "https://trwnh.com/actors/7057bc10-db1c-4ebe-9e00-22cf04be4e5e"
					},
					{
						"rel": "https://webfinger.net/rel/profile-page/",
						"type": "text/html",
						"href": "https://trwnh.com/~a"
					}
				]
			}'''
		}

class JRDResponse(Response):
	media_type = "application/jrd+json"


## Example responses for OpenAPI schema

RESOURCE_NOT_PROVIDED = "Query ?resource={resource} to obtain information about that resource."
RESOURCE_NOT_FOUND = "Resource not found"
RESOURCE_NOT_PARSED = "Resource contains invalid JSON or was unreadable"

RESPONSES = {
	200: {
		"description": "OK: The resource you requested was found, and has the returned resource document.",
		"model": JRD,
		"content": {
			"application/jrd+json": {
				"example": json.loads(JRD.Config.schema_extra['example']),
			},
		},
	},
	400: {
		"description": "Bad Request: No resource was provided.",
		"content": {
			"text/plain": {
				"example": RESOURCE_NOT_PROVIDED
			},
		},
	},
	404: {
		"description": "Not Found: Could not find a document for the provided resource.",
		"content": {
			"text/plain": {
				"example": RESOURCE_NOT_FOUND
			},
		},
	},
	500: {
		"description": "Server Error: The resource exists, but contains invalid JSON or was unreadable.",
		"content": {
			"text/plain": {
				"example": RESOURCE_NOT_PARSED
			},
		},
	},
}


## The lookup method as described in RFC7033

@router.get(
	"",
	summary = "Lookup a Webfinger resource",
	description = "Query the Webfinger service for a resource URI and obtain its associated resource document",
	tags = ["RFC7033"],
	response_model = JRD,
	response_class = JRDResponse,
	responses = {**RESPONSES},
	)
async def lookup(
	resource: str = Query(
		None,
		title = "Resource URI",
		description = "The subject whose document you are looking up. If not provided, you will get a 400 Bad Request."
		),
	rel: list[str] = Query(
		None,
		title = "Link relation",
		description = "If provided, filter the links array for this rel-value only. This parameter may be included multiple times."
		)
	):
	"""
	Respond to a WebFinger query.
	"""
	# If no resource is given, then show a basic hint.
	if not resource:
		return PlainTextResponse(
			content = RESOURCE_NOT_PROVIDED,
			status_code=400,
			headers = HEADERS,
			)
	
	# Otherwise, try to read the resource document.
	try:
		jrd = get_jrd(resource)
	except FileNotFoundError:  # JRD file does not exist
		return PlainTextResponse(
			content = RESOURCE_NOT_FOUND,
			status_code = 404,
			headers = HEADERS,
			)
	except:  # JRD file could not be read or parsed
		return PlainTextResponse(
			content = RESOURCE_NOT_PARSED,
			status_code = 500,
			headers = HEADERS,
			)

	# Obtain values from the resource document.
	subject: str = jrd.get('subject', None) or resource
	aliases: list[str] = [alias for alias in jrd.get('aliases', [])]
	properties: dict[str, str] = jrd.get('properties', {})
	links: list[dict[str,str]] = [link for link in jrd.get('links', [])]

	# 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 and return it.
	content = {
		"subject": subject,
		"aliases": aliases,
		"properties": properties,
		"links": links,
	}
	content = {k: v for k, v in content.items() if v} # remove null values
	return JSONResponse(
		content = content,
		media_type = "application/jrd+json",
		headers = HEADERS,
		)