microblog/app/config.py

291 lines
7.8 KiB
Python
Raw Normal View History

2022-08-13 13:35:39 +00:00
import hashlib
import hmac
2022-06-22 18:11:22 +00:00
import os
2022-07-11 07:34:06 +00:00
import secrets
2022-06-22 18:11:22 +00:00
from pathlib import Path
import bcrypt
2022-07-11 07:34:06 +00:00
import itsdangerous
2022-06-22 18:11:22 +00:00
import pydantic
import tomli
from fastapi import Form
from fastapi import HTTPException
from fastapi import Request
2022-07-11 07:34:06 +00:00
from itsdangerous import URLSafeTimedSerializer
from loguru import logger
2022-10-19 18:46:01 +00:00
from mistletoe import markdown # type: ignore
2022-06-22 18:11:22 +00:00
from app.customization import _CUSTOM_ROUTES
from app.customization import _StreamVisibilityCallback
from app.customization import default_stream_visibility_callback
2022-06-27 18:55:44 +00:00
from app.utils.emoji import _load_emojis
2022-08-24 07:02:20 +00:00
from app.utils.version import get_version_commit
2022-06-27 18:55:44 +00:00
2022-06-22 18:11:22 +00:00
ROOT_DIR = Path().parent.resolve()
2022-07-07 18:37:16 +00:00
_CONFIG_FILE = os.getenv("MICROBLOGPUB_CONFIG_FILE", "profile.toml")
2022-06-22 18:11:22 +00:00
2022-07-18 18:44:55 +00:00
VERSION_COMMIT = "dev"
try:
from app._version import VERSION_COMMIT # type: ignore
except ImportError:
2022-08-24 07:02:20 +00:00
VERSION_COMMIT = get_version_commit()
2022-07-18 18:44:55 +00:00
2022-08-13 13:35:39 +00:00
# Force reloading cache when the CSS is updated
CSS_HASH = "none"
try:
css_data = (ROOT_DIR / "app" / "static" / "css" / "main.css").read_bytes()
CSS_HASH = hashlib.md5(css_data, usedforsecurity=False).hexdigest()
except FileNotFoundError:
pass
2022-08-29 18:11:31 +00:00
# Force reloading cache when the JS is changed
JS_HASH = "none"
try:
# To keep things simple, we keep a single hash for the 2 files
dat = b""
for j in [
ROOT_DIR / "app" / "static" / "common.js",
ROOT_DIR / "app" / "static" / "common-admin.js",
ROOT_DIR / "app" / "static" / "new.js",
]:
dat += j.read_bytes()
JS_HASH = hashlib.md5(dat, usedforsecurity=False).hexdigest()
2022-08-29 18:11:31 +00:00
except FileNotFoundError:
pass
2022-09-08 18:00:02 +00:00
MOVED_TO_FILE = ROOT_DIR / "data" / "moved_to.dat"
def _get_moved_to() -> str | None:
if not MOVED_TO_FILE.exists():
return None
return MOVED_TO_FILE.read_text()
def set_moved_to(moved_to: str) -> None:
MOVED_TO_FILE.write_text(moved_to)
2022-07-04 19:45:23 +00:00
VERSION = f"2.0.0+{VERSION_COMMIT}"
2022-06-22 18:11:22 +00:00
USER_AGENT = f"microblogpub/{VERSION}"
AP_CONTENT_TYPE = "application/activity+json"
class _PrivacyReplace(pydantic.BaseModel):
domain: str
replace_by: str
2022-08-10 06:58:18 +00:00
class _ProfileMetadata(pydantic.BaseModel):
key: str
value: str
2022-08-15 08:15:00 +00:00
class _BlockedServer(pydantic.BaseModel):
hostname: str
reason: str | None = None
2022-06-22 18:11:22 +00:00
class Config(pydantic.BaseModel):
domain: str
username: str
admin_password: bytes
name: str
summary: str
https: bool
icon_url: str | None = None
image_url: str | None = None
2022-06-22 18:11:22 +00:00
secret: str
debug: bool = False
2022-07-30 06:46:29 +00:00
trusted_hosts: list[str] = ["127.0.0.1"]
2023-01-11 22:33:50 +00:00
tags: list[str] | None = None
manually_approves_followers: bool = False
privacy_replace: list[_PrivacyReplace] | None = None
2022-08-10 06:58:18 +00:00
metadata: list[_ProfileMetadata] | None = None
2022-08-04 17:10:57 +00:00
code_highlighting_theme = "friendly_grayscale"
2022-08-15 08:15:00 +00:00
blocked_servers: list[_BlockedServer] = []
2022-08-24 19:18:30 +00:00
custom_footer: str | None = None
2022-08-31 17:16:03 +00:00
emoji: str | None = None
also_known_as: str | None = None
2023-01-12 17:04:19 +00:00
ga_analytics: str | None = None
2022-06-22 18:11:22 +00:00
hides_followers: bool = False
hides_following: bool = False
inbox_retention_days: int = 15
2022-11-09 20:26:43 +00:00
custom_content_security_policy: str | None = None
webfinger_domain: str | None = None
2022-06-22 18:11:22 +00:00
# Config items to make tests easier
2022-06-29 18:43:17 +00:00
sqlalchemy_database: str | None = None
2022-06-22 18:11:22 +00:00
key_path: str | None = None
2022-11-21 19:43:51 +00:00
session_timeout: int = 3600 * 24 * 3 # in seconds, 3 days by default
disabled_notifications: list[str] = []
2022-11-04 18:28:21 +00:00
# Only set when the app is served on a non-root path
id: str | None = None
2022-06-22 18:11:22 +00:00
def load_config() -> Config:
try:
return Config.parse_obj(
tomli.loads((ROOT_DIR / "data" / _CONFIG_FILE).read_text())
)
except FileNotFoundError:
2022-06-22 18:48:48 +00:00
raise ValueError(
f"Please run the configuration wizard, {_CONFIG_FILE} is missing"
)
2022-06-22 18:11:22 +00:00
def is_activitypub_requested(req: Request) -> bool:
accept_value = req.headers.get("accept")
if not accept_value:
return False
for val in {
"application/ld+json",
"application/activity+json",
}:
if accept_value.startswith(val):
return True
return False
def verify_password(pwd: str) -> bool:
return bcrypt.checkpw(pwd.encode(), CONFIG.admin_password)
CONFIG = load_config()
DOMAIN = CONFIG.domain
_SCHEME = "https" if CONFIG.https else "http"
ID = f"{_SCHEME}://{DOMAIN}"
2022-11-04 18:28:21 +00:00
# When running the app on a path, the ID maybe set by the config, but in this
# case, a valid webfinger must be served on the root domain
if CONFIG.id:
ID = CONFIG.id
2022-06-22 18:11:22 +00:00
USERNAME = CONFIG.username
# Allow to use @handle@webfinger-domain.tld while hosting the server at domain.tld
WEBFINGER_DOMAIN = CONFIG.webfinger_domain or DOMAIN
MANUALLY_APPROVES_FOLLOWERS = CONFIG.manually_approves_followers
HIDES_FOLLOWERS = CONFIG.hides_followers
HIDES_FOLLOWING = CONFIG.hides_following
PRIVACY_REPLACE = None
if CONFIG.privacy_replace:
PRIVACY_REPLACE = {pr.domain: pr.replace_by for pr in CONFIG.privacy_replace}
2022-08-15 08:15:00 +00:00
BLOCKED_SERVERS = {blocked_server.hostname for blocked_server in CONFIG.blocked_servers}
ALSO_KNOWN_AS = CONFIG.also_known_as
2022-11-09 20:26:43 +00:00
CUSTOM_CONTENT_SECURITY_POLICY = CONFIG.custom_content_security_policy
2022-08-15 08:15:00 +00:00
INBOX_RETENTION_DAYS = CONFIG.inbox_retention_days
2022-11-21 19:43:51 +00:00
SESSION_TIMEOUT = CONFIG.session_timeout
2022-08-24 19:18:30 +00:00
CUSTOM_FOOTER = (
2022-10-19 18:46:01 +00:00
markdown(CONFIG.custom_footer.replace("{version}", VERSION))
2022-08-24 19:18:30 +00:00
if CONFIG.custom_footer
else None
)
2022-06-22 18:11:22 +00:00
BASE_URL = ID
DEBUG = CONFIG.debug
2022-06-29 18:43:17 +00:00
DB_PATH = CONFIG.sqlalchemy_database or ROOT_DIR / "data" / "microblogpub.db"
SQLALCHEMY_DATABASE_URL = f"sqlite:///{DB_PATH}"
2022-06-22 18:11:22 +00:00
KEY_PATH = (
(ROOT_DIR / CONFIG.key_path) if CONFIG.key_path else ROOT_DIR / "data" / "key.pem"
)
2022-06-27 18:55:44 +00:00
EMOJIS = "😺 😸 😹 😻 😼 😽 🙀 😿 😾"
2022-08-31 17:16:03 +00:00
if CONFIG.emoji:
EMOJIS = CONFIG.emoji
2022-06-27 18:55:44 +00:00
# Emoji template for the FE
2022-11-04 18:28:21 +00:00
EMOJI_TPL = (
'<img src="{base_url}/static/twemoji/{filename}.svg" alt="{raw}" class="emoji">'
)
2022-06-27 18:55:44 +00:00
_load_emojis(ROOT_DIR, BASE_URL)
2022-06-22 18:11:22 +00:00
2022-08-04 17:10:57 +00:00
CODE_HIGHLIGHTING_THEME = CONFIG.code_highlighting_theme
2022-09-08 18:00:02 +00:00
MOVED_TO = _get_moved_to()
2022-06-22 18:11:22 +00:00
_NavBarItem = tuple[str, str]
class NavBarItems:
EXTRA_NAVBAR_ITEMS: list[_NavBarItem] = []
INDEX_NAVBAR_ITEM: _NavBarItem | None = None
NOTES_PATH = "/"
def load_custom_routes() -> None:
try:
from data import custom_routes # type: ignore # noqa: F401
except ImportError:
pass
for path, custom_handler in _CUSTOM_ROUTES.items():
# If a handler wants to replace the root, move the index to /notes
if path == "/":
NavBarItems.NOTES_PATH = "/notes"
NavBarItems.INDEX_NAVBAR_ITEM = (path, custom_handler.title)
else:
if custom_handler.show_in_navbar:
NavBarItems.EXTRA_NAVBAR_ITEMS.append((path, custom_handler.title))
2022-07-11 07:42:39 +00:00
session_serializer = URLSafeTimedSerializer(
CONFIG.secret,
salt=f"{ID}.session",
)
2022-07-11 07:34:06 +00:00
csrf_serializer = URLSafeTimedSerializer(
2022-07-11 07:42:39 +00:00
CONFIG.secret,
salt=f"{ID}.csrf",
2022-06-22 18:11:22 +00:00
)
def generate_csrf_token() -> str:
2022-07-11 07:34:06 +00:00
return csrf_serializer.dumps(secrets.token_hex(16)) # type: ignore
2022-06-22 18:11:22 +00:00
def verify_csrf_token(
csrf_token: str = Form(),
redirect_url: str | None = Form(None),
) -> None:
please_try_again = "please try again"
if redirect_url:
please_try_again = f'<a href="{redirect_url}">please try again</a>'
2022-07-11 07:34:06 +00:00
try:
2022-07-11 07:42:39 +00:00
csrf_serializer.loads(csrf_token, max_age=1800)
2022-07-11 07:34:06 +00:00
except (itsdangerous.BadData, itsdangerous.SignatureExpired):
logger.exception("Failed to verify CSRF token")
raise HTTPException(
status_code=403,
2022-09-16 16:37:09 +00:00
detail=f"The security token has expired, {please_try_again}",
)
2022-06-22 18:11:22 +00:00
return None
2022-11-18 05:35:04 +00:00
def hmac_sha256() -> hmac.HMAC:
return hmac.new(CONFIG.secret.encode(), digestmod=hashlib.sha256)
stream_visibility_callback: _StreamVisibilityCallback
try:
from data.stream import ( # type: ignore # noqa: F401, E501
custom_stream_visibility_callback,
)
stream_visibility_callback = custom_stream_visibility_callback
except ImportError:
stream_visibility_callback = default_stream_visibility_callback