move config parsing to own module

docker-compose.staging.yaml

@@ -12,7 +12,7 @@ services:
       ENCRYPTION_KEY: bi5FDwhZGKfc4urLJ_ChGtIAaOPgxd3RDOhnvct10mw=
       SECRET_KEY: cb3f4afde364bfb3956b97ca22ef4d2b593d9d980a4330686267cabcd2c0befd
       SQLALCHEMY_DATABASE_URI: postgresql://hushline:hushline@postgres:5432/hushline
-      REGISTRATION_CODES_REQUIRED: False
+      REGISTRATION_CODES_REQUIRED: "false"
       SESSION_COOKIE_NAME: session
       NOTIFICATIONS_ADDRESS: notifications@hushline.app
       DIRECTORY_VERIFIED_TAB_ENABLED: "${DIRECTORY_VERIFIED_TAB_ENABLED:-true}"

docker-compose.stripe.yaml

@@ -12,7 +12,7 @@ services:
       ENCRYPTION_KEY: bi5FDwhZGKfc4urLJ_ChGtIAaOPgxd3RDOhnvct10mw=
       SECRET_KEY: cb3f4afde364bfb3956b97ca22ef4d2b593d9d980a4330686267cabcd2c0befd
       SQLALCHEMY_DATABASE_URI: postgresql://hushline:hushline@postgres:5432/hushline
-      REGISTRATION_CODES_REQUIRED: False
+      REGISTRATION_CODES_REQUIRED: "false"
       SESSION_COOKIE_NAME: session
       NOTIFICATIONS_ADDRESS: notifications@hushline.app
       DIRECTORY_VERIFIED_TAB_ENABLED: "${DIRECTORY_VERIFIED_TAB_ENABLED:-true}"

hushline/__init__.py

@@ -1,124 +1,50 @@
 import asyncio
 import logging
-import os
-from datetime import timedelta
-from typing import Any
+from typing import Any, Mapping, Optional
 
 from flask import Flask, flash, redirect, request, session, url_for
 from flask.cli import AppGroup
 from jinja2 import StrictUndefined
-from markupsafe import Markup
 from werkzeug.wrappers.response import Response
 
 from . import admin, premium, routes, settings
+from .config import load_config
 from .db import db, migrate
 from .model import HostOrganization, Tier, User
 from .version import __version__
 
 
-def create_app() -> Flask:
+def create_app(config: Optional[Mapping[str, Any]] = None) -> Flask:
     app = Flask(__name__)
-
-    # Configure logging
     app.logger.setLevel(logging.DEBUG)
 
-    # sqlalchemy configs
-    app.config["SQLALCHEMY_DATABASE_URI"] = os.environ.get("SQLALCHEMY_DATABASE_URI")
-    # if it's a Postgres URI, replace the scheme with `postgresql+psycopg`
-    # because we're using the psycopg driver
-    if app.config["SQLALCHEMY_DATABASE_URI"].startswith("postgresql://"):
-        app.config["SQLALCHEMY_DATABASE_URI"] = app.config["SQLALCHEMY_DATABASE_URI"].replace(
-            "postgresql://", "postgresql+psycopg://", 1
-        )
-    app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
-
-    # cookie configs
-    app.config["FLASK_ENV"] = os.environ.get("FLASK_ENV")
-    app.config["SECRET_KEY"] = os.environ.get("SECRET_KEY")
-    app.config["SESSION_COOKIE_NAME"] = os.environ.get("SESSION_COOKIE_NAME", "__HOST-session")
-    app.config["SESSION_COOKIE_SECURE"] = True
-    app.config["SESSION_COOKIE_HTTPONLY"] = True
-    app.config["SESSION_COOKIE_SAMESITE"] = "Lax"
-    app.config["PERMANENT_SESSION_LIFETIME"] = timedelta(minutes=30)
+    if not config:
+        config = load_config()
 
     # hushline specific configs
-    app.config["VERSION"] = __version__
-    app.config["ENCRYPTION_KEY"] = os.getenv("ENCRYPTION_KEY")
-    app.config["ONION_HOSTNAME"] = os.environ.get("ONION_HOSTNAME", None)
-    app.config["DIRECTORY_VERIFIED_TAB_ENABLED"] = (
-        os.environ.get("DIRECTORY_VERIFIED_TAB_ENABLED", "true").lower() == "true"
-    )
-    app.config["SMTP_FORWARDING_MESSAGE_HTML"] = (
-        Markup(os.environ.get("SMTP_FORWARDING_MESSAGE_HTML", "")) or None
-    )
-    app.config["REGISTRATION_CODES_REQUIRED"] = (
-        os.environ.get("REGISTRATION_CODES_REQUIRED", "true").lower() == "true"
-    )
-    app.config["NOTIFICATIONS_ADDRESS"] = os.environ.get("NOTIFICATIONS_ADDRESS", None)
-    app.config["SMTP_USERNAME"] = os.environ.get("SMTP_USERNAME", None)
-    app.config["SMTP_PASSWORD"] = os.environ.get("SMTP_PASSWORD", None)
-    app.config["SMTP_SERVER"] = os.environ.get("SMTP_SERVER", None)
-    app.config["SMTP_PORT"] = int(os.environ.get("SMTP_PORT", 0))
-    app.config["SMTP_ENCRYPTION"] = os.environ.get("SMTP_ENCRYPTION", "StartTLS")
-    app.config["REQUIRE_PGP"] = os.environ.get("REQUIRE_PGP", "False").lower() == "true"
-    app.config["STRIPE_PUBLISHABLE_KEY"] = os.environ.get("STRIPE_PUBLISHABLE_KEY", None)
-    app.config["STRIPE_SECRET_KEY"] = os.environ.get("STRIPE_SECRET_KEY", None)
-    app.config["STRIPE_WEBHOOK_SECRET"] = os.environ.get("STRIPE_WEBHOOK_SECRET", None)
-
-    # Handle the tips domain for profile verification
-    app.config["SERVER_NAME"] = os.getenv("SERVER_NAME")
-    app.config["PREFERRED_URL_SCHEME"] = "https" if os.getenv("SERVER_NAME") is not None else "http"
-
-    # jinja configs
-    app.jinja_env.globals["hushline_version"] = __version__
-    app.jinja_env.globals["directory_verified_tab_enabled"] = app.config[
-        "DIRECTORY_VERIFIED_TAB_ENABLED"
-    ]
-    if app.config.get("FLASK_ENV", None) == "development":
-        app.logger.info("Development environment detected, enabling jinja2.StrictUndefined")
-        app.jinja_env.undefined = StrictUndefined
-
-    # always pop the config to avoid accidentally dumping all our secrets to the user
-    app.jinja_env.globals.pop("config", None)
-    app.jinja_env.globals["smtp_forwarding_message_html"] = app.config[
-        "SMTP_FORWARDING_MESSAGE_HTML"
-    ]
-    if onion_hostname := app.config.get("ONION_HOSTNAME", None):
-        app.jinja_env.globals["onion_hostname"] = onion_hostname
 
+    app.config.from_mapping(config)
+    configure_jinja(app)
     db.init_app(app)
     migrate.init_app(app, db)
 
-    # Initialize Stripe
-    if app.config["STRIPE_SECRET_KEY"]:
-        with app.app_context():
-            premium.init_stripe()
-
     routes.init_app(app)
     for module in [admin, settings]:
         app.register_blueprint(module.create_blueprint())
 
-    if app.config["STRIPE_SECRET_KEY"]:
+    if app.config.get("STRIPE_SECRET_KEY"):
         app.register_blueprint(premium.create_blueprint(app))
+        # Initialize Stripe
+        with app.app_context():
+            premium.init_stripe()
 
     @app.errorhandler(404)
     def page_not_found(e: Exception) -> Response:
         flash("⛓️‍💥 That page doesn't exist.", "warning")
         return redirect(url_for("index"))
 
-    @app.context_processor
-    def inject_variables() -> dict[str, Any]:
-        data = {
-            "host_org": HostOrganization.fetch_or_default(),
-            "is_premium_enabled": bool(app.config.get("STRIPE_SECRET_KEY", False)),
-            "is_onion_service": request.host.lower().endswith(".onion"),
-        }
-        if "user_id" in session:
-            data["user"] = db.session.get(User, session["user_id"])
-        return data
-
     # Add Onion-Location header to all responses
-    if app.config["ONION_HOSTNAME"]:
+    if app.config.get("ONION_HOSTNAME"):
 
         @app.after_request
         def add_onion_location_header(response: Response) -> Response:
@@ -133,6 +59,35 @@ def add_onion_location_header(response: Response) -> Response:
     return app
 
 
+def configure_jinja(app: Flask) -> None:
+    app.jinja_env.globals["hushline_version"] = __version__
+    app.jinja_env.globals["directory_verified_tab_enabled"] = app.config[
+        "DIRECTORY_VERIFIED_TAB_ENABLED"
+    ]
+    if app.config.get("FLASK_ENV") == "development":
+        app.logger.info("Development environment detected, enabling jinja2.StrictUndefined")
+        app.jinja_env.undefined = StrictUndefined
+
+    # always pop the config to avoid accidentally dumping all our secrets to the user
+    app.jinja_env.globals.pop("config", None)
+    app.jinja_env.globals["smtp_forwarding_message_html"] = app.config[
+        "SMTP_FORWARDING_MESSAGE_HTML"
+    ]
+    if onion_hostname := app.config.get("ONION_HOSTNAME"):
+        app.jinja_env.globals["onion_hostname"] = onion_hostname
+
+    @app.context_processor
+    def inject_variables() -> dict[str, Any]:
+        data = {
+            "host_org": HostOrganization.fetch_or_default(),
+            "is_premium_enabled": bool(app.config.get("STRIPE_SECRET_KEY", False)),
+            "is_onion_service": request.host.lower().endswith(".onion"),
+        }
+        if "user_id" in session:
+            data["user"] = db.session.get(User, session["user_id"])
+        return data
+
+
 def register_commands(app: Flask) -> None:
     stripe_cli = AppGroup("stripe")
 

hushline/config.py

@@ -0,0 +1,100 @@
+import os
+from datetime import timedelta
+from typing import Any, Mapping, Optional
+
+from markupsafe import Markup
+
+from .utils import if_not_none, parse_bool
+
+
+def load_config(env: Optional[Mapping[str, str]] = None) -> Mapping[str, Any]:
+    if env is None:
+        env = os.environ
+
+    config: dict[str, Any] = {}
+    for func in [_load_flask, _load_sqlalchemy, _load_smtp, _load_stripe, _load_hushline_misc]:
+        config |= func()
+
+    return config
+
+
+def _load_flask() -> Mapping[str, Any]:
+    data = {
+        "SESSION_COOKIE_NAME": os.environ.get("SESSION_COOKIE_NAME", "__HOST-session"),
+        "SESSION_COOKIE_SECURE": True,
+        "SESSION_COOKIE_HTTPONLY": True,
+        "SESSION_COOKIE_SAMESITE": "Lax",
+        "PERMANENT_SESSION_LIFETIME": timedelta(minutes=30),
+    }
+
+    # Handle the tips domain for profile verification
+    if server_name := os.environ.get("SERVER_NAME"):
+        data["SERVER_NAME"] = server_name
+    data["PREFERRED_URL_SCHEME"] = "https" if server_name else "http"
+
+    for key in ["FLASK_ENV", "SECRET_KEY"]:
+        if val := os.environ.get(key):
+            data[key] = val
+
+    return data
+
+
+def _load_sqlalchemy() -> Mapping[str, Any]:
+    data: dict[str, Any] = {}
+
+    if db_uri := os.environ.get("SQLALCHEMY_DATABASE_URI"):
+        # if it's a Postgres URI, replace the scheme with `postgresql+psycopg`
+        # because we're using the psycopg driver
+        if db_uri.startswith("postgresql://"):
+            db_uri = db_uri.replace("postgresql://", "postgresql+psycopg://", 1)
+        data["SQLALCHEMY_DATABASE_URI"] = db_uri
+
+    data["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
+
+    return data
+
+
+def _load_smtp() -> Mapping[str, Any]:
+    data: dict[str, Any] = {}
+
+    for key in ["NOTIFICATIONS_ADDRESS", "SMTP_USERNAME", "SMTP_PASSWORD", "SMTP_SERVER"]:
+        if val := os.environ.get(key):
+            data[key] = val
+
+    data["SMTP_PORT"] = if_not_none(os.environ.get("SMTP_PORT"), int)
+    data["SMTP_ENCRYPTION"] = os.environ.get("SMTP_ENCRYPTION", "StartTLS")
+    data["SMTP_FORWARDING_MESSAGE_HTML"] = if_not_none(
+        os.environ.get("SMTP_FORWARDING_MESSAGE_HTML"), Markup, allow_falsey=False
+    )
+
+    return data
+
+
+def _load_hushline_misc() -> Mapping[str, Any]:
+    data: dict[str, Any] = {
+        "ENCRYPTION_KEY": os.environ["ENCRYPTION_KEY"],
+    }
+
+    if onion := os.environ.get("ONION_HOSTNAME"):
+        data["ONION_HOSTNAME"] = onion
+
+    for key, default in [
+        ("DIRECTORY_VERIFIED_TAB_ENABLED", True),
+        ("REGISTRATION_CODES_REQUIRED", True),
+        ("REQUIRE_PGP", False),
+    ]:
+        if value := os.environ.get(key):
+            data[key] = parse_bool(value)
+        else:
+            data[key] = default
+    return data
+
+
+def _load_stripe() -> Mapping[str, Any]:
+    data = {}
+
+    for key in ["STRIPE_PUBLISHABLE_KEY", "STRIPE_SECRET_KEY", "STRIPE_WEBHOOK_SECRET"]:
+        if value := os.environ.get(key):
+            data[key] = value
+
+    return data

hushline/utils.py

@@ -4,14 +4,37 @@
 from email.mime.multipart import MIMEMultipart
 from email.mime.text import MIMEText
 from functools import wraps
-from typing import Any, Callable, Generator
+from typing import Any, Callable, Generator, Optional, TypeVar
 
 from flask import abort, current_app, flash, redirect, session, url_for
 
 from hushline.model import SMTPEncryption, User
 
 from .db import db
 
+T = TypeVar("T")
+U = TypeVar("U")
+
+
+def if_not_none(
+    value: Optional[T], func: Callable[[T], U], allow_falsey: bool = True
+) -> Optional[U]:
+    if allow_falsey:
+        if value is not None:
+            return func(value)
+    elif value:
+        return func(value)
+    return None
+
+
+def parse_bool(val: str) -> bool:
+    match val:
+        case "true":
+            return True
+        case "false":
+            return False
+    raise ValueError(f"Unparseable boolean value: {val!r}")
+
 
 def authentication_required(f: Callable[..., Any]) -> Callable[..., Any]:
     @wraps(f)

tests/conftest.py

@@ -172,7 +172,7 @@ def env_var_modifier(mocker: MockFixture) -> Callable[[MockFixture], None]:
 def app(
     database: str, mocker: MockFixture, env_var_modifier: Callable[[MockFixture], None]
 ) -> Generator[Flask, None, None]:
-    os.environ["REGISTRATION_CODES_REQUIRED"] = "False"
+    os.environ["REGISTRATION_CODES_REQUIRED"] = "false"
     os.environ["SQLALCHEMY_DATABASE_URI"] = CONN_FMT_STR.format(database=database)
     os.environ["STRIPE_SECRET_KEY"] = "sk_test_123"  # For premium tests