Support canonical subject

Also support resource URIs with slashes

Misc: Better OpenAPI examples and docstrings
This commit is contained in:
a 2022-12-14 22:19:32 -06:00
parent 8cd70fca10
commit 50d9e4bf6f
4 changed files with 108 additions and 67 deletions

View file

@ -16,4 +16,6 @@ license = {text = "AGPL"}
requires = ["pdm-pep517>=1.0.0"]
build-backend = "pdm.pep517.api"
[tool.pdm]
[tool.pdm.scripts]
dev = "uvicorn webfinger:app --port 7033 --reload"
start = "python -m webfinger"

View file

@ -3,7 +3,7 @@ import webfinger.lookup
app = FastAPI(
title = "webfinger",
version = "2"
version = "1"
)
app.include_router(webfinger.lookup.router)

View file

@ -1,19 +1,44 @@
import json
from os import environ as env
from urllib.parse import quote_plus as url_encode
from pathlib import Path
def get_document(resource: str) -> dict | None:
def get_jrd(resource: str) -> dict[str, str | list[str] | dict[str,str]]:
"""
Obtain a JSON Resource Descriptor (JRD)
A JRD is a JSON object containing the following:
- subject (string value; SHOULD be present)
- aliases (array of string values; OPTIONAL)
- properties (object containing key-value pairs as strings; OPTIONAL)
- links (array of objects containing link relation information; OPTIONAL)
Parameters:
resource (str): The URI of the resouce
Returns:
jrd (dict): Parsed JRD
Raises:
FileNotFoundError: No JRD file exists in the resource directory
OSError: A file may exist, but it is not readable
json.JSONDecodeError: A file exists, but does not contain a valid JSON object
"""
# Get a filename for the resource document.
dir = env.get("RESOURCE_DIR") or "resource"
filename = f"{dir}/{resource}.json"
filename = resource
path = f"{dir}/{filename}.json"
# Check that the file exists.
path = Path(filename)
if not path:
return None
# If we can't get a file, try percent-encoding
if not Path(path).is_file():
filename = url_encode(resource)
path = f"{dir}/{filename}.json"
# Open the file and load the JSON as a dictionary.
with open(filename, "r") as file:
data: dict = json.loads(file.read())
try:
with open(path, "r") as file:
jrd: dict[str, str | list[str] | dict[str,str]] = json.loads(file.read())
except:
raise
return data
return jrd

View file

@ -3,8 +3,9 @@ from fastapi.responses import PlainTextResponse, JSONResponse
from fastapi.params import Query
from pydantic import BaseModel
import json
from webfinger.io import get_document
from webfinger.io import get_jrd
router = APIRouter(
@ -16,26 +17,6 @@ router = APIRouter(
MEDIA_TYPE = "application/jrd+json"
EXAMPLE_RESOURCE = r'''
{
"aliases": ["https://trwnh.com/actors/7057bc10-db1c-4ebe-9e00-22cf04be4e5e", "https://trwnh.com/~a"],
"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"
}
]
}
'''
NO_RESOURCE_HINT = "Query ?resource={resource} to obtain information about that resource."
## Pydantic models for OpenAPI schema
@ -48,54 +29,81 @@ class Link(BaseModel):
class Config:
schema_extra = {
"example": '''{
"example": r'''{
"rel": "https://webfinger.net/rel/profile-page/",
"type": "text/html",
"href": "https://trwnh.com/~a"
}'''
}
class Resource(BaseModel):
class JRD(BaseModel):
subject: str
aliases: list[str] | None
links: list[Link] | None
class Config:
schema_extra = {
"example": EXAMPLE_RESOURCE
"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 WebfingerResponse(Response):
class JRDResponse(Response):
media_type = MEDIA_TYPE
## 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": {
MEDIA_TYPE: {
"example": json.loads(JRD.Config.schema_extra['example']),
},
},
},
400: {
"description": "Bad Request. No resource was provided.",
"description": "Bad Request: No resource was provided.",
"content": {
"text/plain": {
"example": NO_RESOURCE_HINT
}
"example": RESOURCE_NOT_PROVIDED
},
},
},
404: {
"description": "Not Found. Could not find a document for the provided resource.",
"description": "Not Found: Could not find a document for the provided resource.",
"content": {
"text/plain": {
"example": "{resource} not found"
}
"example": RESOURCE_NOT_FOUND
},
},
200: {
"description": "The resource you requested was found, and has the returned resource document.",
"model": Resource,
},
500: {
"description": "Server Error: The resource exists, but contains invalid JSON or was unreadable.",
"content": {
MEDIA_TYPE: {
"example": EXAMPLE_RESOURCE
}
}
"text/plain": {
"example": RESOURCE_NOT_PARSED
},
},
},
}
@ -107,46 +115,52 @@ RESPONSES = {
summary = "Lookup a Webfinger resource",
description = "Query the Webfinger service for a resource URI and obtain its associated resource document",
tags = ["RFC7033"],
response_model = Resource,
response_class = WebfingerResponse,
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=NO_RESOURCE_HINT,
content=RESOURCE_NOT_PROVIDED,
status_code=400,
)
# Otherwise, try to read the resource document.
data = get_document(resource)
# If the document could not be read, then it likely didn't exist.
if data is None:
try:
jrd = get_jrd(resource)
except FileNotFoundError: # JRD file does not exist
return PlainTextResponse(
content=f"{resource} not found",
content=RESOURCE_NOT_FOUND,
status_code=404,
)
except: # JRD file could not be read or parsed
return PlainTextResponse(
content=RESOURCE_NOT_PARSED,
status_code=500,
)
# Obtain values from the resource document.
aliases = [alias for alias in data.get('aliases', [])]
links = [link for link in data.get('links', [])]
properties = data.get('properties', {})
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:
@ -154,7 +168,7 @@ async def lookup(
# Construct JSON response and return it.
content = {
"subject": resource,
"subject": subject,
"aliases": aliases,
"properties": properties,
"links": links,