forked from forks/microblog.pub
Tweak the webfinger resolution
This commit is contained in:
parent
d90e489fc6
commit
06f4f824d8
4 changed files with 52 additions and 30 deletions
|
@ -28,6 +28,8 @@
|
|||
- Privacy-aware image upload endpoint that strip EXIF meta data before storing the file
|
||||
- No JavaScript, that's it, even the admin UI is pure HTML/CSS
|
||||
- Easy to customize (the theme is written Sass)
|
||||
- mobile-friendly theme
|
||||
- with dark and light version
|
||||
- Microformats aware (exports `h-feed`, `h-entry`, `h-cards`, ...)
|
||||
- Exports RSS/Atom feeds
|
||||
- Comes with a tiny HTTP API to help posting new content and performing basic actions
|
||||
|
|
|
@ -44,6 +44,7 @@ with open('config/me.yml') as f:
|
|||
ICON_URL = conf['icon_url']
|
||||
PASS = conf['pass']
|
||||
PUBLIC_INSTANCES = conf.get('public_instances')
|
||||
# TODO(tsileo): choose dark/light style
|
||||
THEME_COLOR = conf.get('theme_color')
|
||||
|
||||
USER_AGENT = (
|
||||
|
|
|
@ -9,6 +9,10 @@ from . import strtobool
|
|||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class InvalidURLError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def is_url_valid(url: str) -> bool:
|
||||
parsed = urlparse(url)
|
||||
if parsed.scheme not in ['http', 'https']:
|
||||
|
@ -33,3 +37,10 @@ def is_url_valid(url: str) -> bool:
|
|||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def check_url(url: str) -> None:
|
||||
if not is_url_valid(url):
|
||||
raise InvalidURLError(f'"{url}" is invalid')
|
||||
|
||||
return None
|
||||
|
|
|
@ -1,15 +1,25 @@
|
|||
from typing import Optional
|
||||
from urllib.parse import urlparse
|
||||
from typing import Dict, Any
|
||||
from typing import Optional
|
||||
import logging
|
||||
|
||||
import requests
|
||||
|
||||
def get_remote_follow_template(resource: str) -> Optional[str]:
|
||||
"""Mastodon-like WebFinger resolution to retrieve the activity stream Actor URL.
|
||||
from .urlutils import check_url
|
||||
|
||||
Returns:
|
||||
the Actor URL or None if the resolution failed.
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def webfinger(resource: str) -> Optional[Dict[str, Any]]:
|
||||
"""Mastodon-like WebFinger resolution to retrieve the activity stream Actor URL.
|
||||
"""
|
||||
if resource.startswith('http'):
|
||||
logger.info(f'performing webfinger resolution for {resource}')
|
||||
protos = ['https', 'http']
|
||||
if resource.startswith('http://'):
|
||||
protos.reverse()
|
||||
host = urlparse(resource).netloc
|
||||
elif resource.startswith('https://'):
|
||||
host = urlparse(resource).netloc
|
||||
else:
|
||||
if resource.startswith('acct:'):
|
||||
|
@ -18,15 +28,30 @@ def get_remote_follow_template(resource: str) -> Optional[str]:
|
|||
resource = resource[1:]
|
||||
_, host = resource.split('@', 1)
|
||||
resource='acct:'+resource
|
||||
resp = requests.get(
|
||||
f'https://{host}/.well-known/webfinger',
|
||||
{'resource': resource}
|
||||
)
|
||||
print(resp, resp.request.url)
|
||||
|
||||
# Security check on the url (like not calling localhost)
|
||||
check_url(f'https://{host}')
|
||||
|
||||
for i, proto in enumerate(protos):
|
||||
try:
|
||||
url = f'{proto}://{host}/.well-known/webfinger'
|
||||
resp = requests.get(
|
||||
url,
|
||||
{'resource': resource}
|
||||
)
|
||||
except requests.ConnectionError:
|
||||
# If we tried https first and the domain is "http only"
|
||||
if i == 0:
|
||||
continue
|
||||
break
|
||||
if resp.status_code == 404:
|
||||
return None
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
return resp.json()
|
||||
|
||||
|
||||
def get_remote_follow_template(resource: str) -> Optional[str]:
|
||||
data = webfinger(resource)
|
||||
for link in data['links']:
|
||||
if link.get('rel') == 'http://ostatus.org/schema/1.0/subscribe':
|
||||
return link.get('template')
|
||||
|
@ -39,24 +64,7 @@ def get_actor_url(resource: str) -> Optional[str]:
|
|||
Returns:
|
||||
the Actor URL or None if the resolution failed.
|
||||
"""
|
||||
if resource.startswith('http'):
|
||||
host = urlparse(resource).netloc
|
||||
else:
|
||||
if resource.startswith('acct:'):
|
||||
resource = resource[5:]
|
||||
if resource.startswith('@'):
|
||||
resource = resource[1:]
|
||||
_, host = resource.split('@', 1)
|
||||
resource='acct:'+resource
|
||||
resp = requests.get(
|
||||
f'https://{host}/.well-known/webfinger',
|
||||
{'resource': resource}
|
||||
)
|
||||
print(resp, resp.request.url)
|
||||
if resp.status_code == 404:
|
||||
return None
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
data = webfinger(resource)
|
||||
for link in data['links']:
|
||||
if link.get('rel') == 'self' and link.get('type') == 'application/activity+json':
|
||||
return link.get('href')
|
||||
|
|
Loading…
Reference in a new issue