Customer branding 533 v2

Makefile

@@ -27,13 +27,13 @@ migrate-prod: ## Run prod env (alembic) migrations
 lint: ## Lint the code
 	poetry run ruff format --check && \
 	poetry run ruff check && \
-	poetry run mypy .
+	poetry run mypy . && \
 	docker compose run --rm app npx prettier --check ./*.md ./docs ./.github/workflows/* ./hushline
 
 .PHONY: fix
 fix: ## Format the code
 	poetry run ruff format && \
-	poetry run ruff check --fix
+	poetry run ruff check --fix && \
 	docker compose run --rm app npx prettier --write ./*.md ./docs ./.github/workflows/* ./hushline
 
 .PHONY: revision

docker-compose.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
     volumes:

hushline/__init__.py

@@ -6,12 +6,13 @@
 from flask import Flask, flash, redirect, request, session, url_for
 from flask_migrate import Migrate
 from jinja2 import StrictUndefined
+from sqlalchemy.exc import ProgrammingError
 from werkzeug.middleware.proxy_fix import ProxyFix
 from werkzeug.wrappers.response import Response
 
 from . import admin, routes, settings
 from .db import db
-from .model import User
+from .model import HostOrganization, User
 from .version import __version__
 
 
@@ -47,6 +48,9 @@ def create_app() -> Flask:
     app.config["IS_PERSONAL_SERVER"] = (
         os.environ.get("IS_PERSONAL_SERVER", "False").lower() == "true"
     )
+    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_SERVER"] = os.environ.get("SMTP_SERVER", None)
@@ -90,6 +94,10 @@ def inject_user() -> dict[str, Any]:
             return {"user": user}
         return {}
 
+    @app.context_processor
+    def inject_host() -> dict[str, HostOrganization]:
+        return dict(host_org=HostOrganization.fetch_or_default())
+
     @app.context_processor
     def inject_is_personal_server() -> dict[str, Any]:
         return {"is_personal_server": app.config["IS_PERSONAL_SERVER"]}
@@ -104,4 +112,17 @@ def add_onion_location_header(response: Response) -> Response:
             )
             return response
 
+    # we can't
+    if app.config.get("FLASK_ENV", None) != "development":
+        with app.app_context():
+            try:
+                host_org = HostOrganization.fetch()
+            except ProgrammingError:
+                app.logger.warning(
+                    "Could not check for existence of HostOrganization", exc_info=True
+                )
+            else:
+                if host_org is None:
+                    app.logger.warning("HostOrganization data not found in database.")
+
     return app

hushline/forms.py

@@ -1,3 +1,4 @@
+import html
 import re
 
 from flask_wtf import FlaskForm
@@ -25,5 +26,22 @@ def __call__(self, form: Form, field: Field) -> None:
             raise ValidationError(self.message)
 
 
+class HexColor:
+    # HTML input color elements only give & accept 6-hexit color codes
+    hex_color_regex: re.Pattern = re.compile(r"^#[0-9a-fA-F]{6}$")
+
+    def __call__(self, form: Form, field: Field) -> None:
+        color: str = field.data
+        if not self.hex_color_regex.match(color):
+            raise ValidationError(f"{color=} is an invalid 6-hexit color code. (eg. #7d25c1)")
+
+
+class CanonicalHTML:
+    def __call__(self, form: Form, field: Field) -> None:
+        text: str = field.data
+        if text != html.escape(text).strip():
+            raise ValidationError(f"{text=} is ambiguous or unescaped.")
+
+
 class TwoFactorForm(FlaskForm):
     verification_code = StringField("2FA Code", validators=[DataRequired(), Length(min=6, max=6)])

hushline/model.py

@@ -2,7 +2,7 @@
 import secrets
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
-from typing import TYPE_CHECKING, Any, Generator, Optional, Sequence
+from typing import TYPE_CHECKING, Any, Generator, Optional, Self, Sequence
 
 from flask_sqlalchemy.model import Model
 from passlib.hash import scrypt
@@ -28,6 +28,46 @@ def default(cls) -> "SMTPEncryption":
         return cls.StartTLS
 
 
+class HostOrganization(Model):
+    __tablename__ = "host_organization"
+
+    _DEFAULT_ID: int = 1
+    _DEFAULT_BRAND_PRIMARY_HEX_COLOR: str = "#7d25c1"
+    _DEFAULT_BRAND_APP_NAME: str = "🤫 Hush Line"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    brand_app_name: Mapped[str] = mapped_column(db.String(255), default=_DEFAULT_BRAND_APP_NAME)
+    brand_primary_hex_color: Mapped[str] = mapped_column(
+        db.String(7), default=_DEFAULT_BRAND_PRIMARY_HEX_COLOR
+    )
+
+    @classmethod
+    def fetch(cls) -> Self | None:
+        return db.session.get(cls, cls._DEFAULT_ID)
+
+    @classmethod
+    def fetch_or_default(cls) -> Self:
+        return cls.fetch() or cls()
+
+    def __init__(self, **kwargs: Any) -> None:
+        # never allow setting of the ID. It should only ever be `1`
+        if "id" in kwargs:
+            raise ValueError(f"Cannot manually set {self.__class__.__name__} attribute `id`")
+
+        # always initialize all values so that the object is populated for use in templates
+        # even when it's not pulled from the DB.
+        # yes, we have to do this here and not rely on `mapped_column(default='...')` because
+        # that logic doesn't trigger until insert, and we want these here pre-insert
+        if "brand_app_name" not in kwargs:
+            kwargs["brand_app_name"] = self._DEFAULT_BRAND_APP_NAME
+        if "brand_primary_hex_color" not in kwargs:
+            kwargs["brand_primary_hex_color"] = self._DEFAULT_BRAND_PRIMARY_HEX_COLOR
+        super().__init__(
+            id=self._DEFAULT_ID,  # type: ignore[call-arg]
+            **kwargs,
+        )
+
+
 @dataclass(frozen=True, repr=False, eq=False)
 class ExtraField:
     label: Optional[str]
@@ -73,9 +113,11 @@ def __init__(
         is_primary: bool,
         **kwargs: Any,
     ) -> None:
-        super().__init__(**kwargs)
-        self._username = _username
-        self.is_primary = is_primary
+        super().__init__(
+            _username=_username,  # type: ignore[call-arg]
+            is_primary=is_primary,  # type: ignore[call-arg]
+            **kwargs,
+        )
 
     @property
     def username(self) -> str:
@@ -219,7 +261,7 @@ def __init__(self, **kwargs: Any) -> None:
             if key in kwargs:
                 raise ValueError(f"Key {key!r} cannot be mannually set. Try 'password' instead.")
         pw = kwargs.pop("password", None)
-        super().__init__()
+        super().__init__(**kwargs)
         self.password_hash = pw
 
 
@@ -255,11 +297,12 @@ def __init__(
         otp_code: str | None = None,
         timecode: int | None = None,
     ) -> None:
-        super().__init__()
-        self.user_id = user_id
-        self.successful = successful
-        self.otp_code = otp_code
-        self.timecode = timecode
+        super().__init__(
+            user_id=user_id,  # type: ignore[call-arg]
+            successful=successful,  # type: ignore[call-arg]
+            otp_code=otp_code,  # type: ignore[call-arg]
+            timecode=timecode,  # type: ignore[call-arg]
+        )
 
 
 class Message(Model):
@@ -273,8 +316,10 @@ class Message(Model):
     def __init__(self, content: str, **kwargs: Any) -> None:
         if "_content" in kwargs:
             raise ValueError("Cannot set '_content' directly. Use 'content'")
-        super().__init__(**kwargs)
-        self.content = content
+        super().__init__(
+            content=content,  # type: ignore[call-arg]
+            **kwargs,
+        )
 
     @property
     def content(self) -> str | None:
@@ -297,9 +342,10 @@ class InviteCode(Model):
     expiration_date: Mapped[datetime]
 
     def __init__(self) -> None:
-        super().__init__()
-        self.code = secrets.token_urlsafe(16)
-        self.expiration_date = datetime.now(timezone.utc) + timedelta(days=365)
+        super().__init__(
+            code=secrets.token_urlsafe(16),  # type: ignore[call-arg]
+            expiration_date=datetime.now(timezone.utc) + timedelta(days=365),  # type: ignore[call-arg]
+        )
 
     def __repr__(self) -> str:
         return f"<InviteCode {self.code}>"

hushline/routes.py

@@ -1,5 +1,4 @@
 import logging
-import os
 import re
 import secrets
 import socket
@@ -320,7 +319,7 @@ def register() -> Response | str | tuple[Response | str, int]:
             flash("👉 You are already logged in.")
             return redirect(url_for("inbox"))
 
-        require_invite_code = os.environ.get("REGISTRATION_CODES_REQUIRED", "True") == "True"
+        require_invite_code = app.config["REGISTRATION_CODES_REQUIRED"]
         form = RegistrationForm()
         if not require_invite_code:
             del form.invite_code

hushline/settings/__init__.py

@@ -25,8 +25,12 @@
 from ..crypto import is_valid_pgp_key
 from ..db import db
 from ..forms import TwoFactorForm
-from ..model import Message, SMTPEncryption, User, Username
-from ..utils import authentication_required, create_smtp_config
+from ..model import HostOrganization, Message, SMTPEncryption, User, Username
+from ..utils import (
+    admin_authentication_required,
+    authentication_required,
+    create_smtp_config,
+)
 from .forms import (
     ChangePasswordForm,
     ChangeUsernameForm,
@@ -37,6 +41,8 @@
     PGPKeyForm,
     PGPProtonForm,
     ProfileForm,
+    UpdateBrandAppNameForm,
+    UpdateBrandPrimaryColorForm,
 )
 
 
@@ -215,6 +221,8 @@ async def index() -> str | Response:
         directory_visibility_form = DirectoryVisibilityForm()
         new_alias_form = NewAliasForm()
         profile_form = ProfileForm()
+        update_brand_primary_color_form = UpdateBrandPrimaryColorForm()
+        update_brand_app_name_form = UpdateBrandAppNameForm()
 
         if request.method == "POST":
             if "update_bio" in request.form and profile_form.validate_on_submit():
@@ -295,6 +303,8 @@ async def index() -> str | Response:
             "settings/index.html",
             user=user,
             all_users=all_users,
+            update_brand_primary_color_form=update_brand_primary_color_form,
+            update_brand_app_name_form=update_brand_app_name_form,
             email_forwarding_form=email_forwarding_form,
             change_password_form=change_password_form,
             change_username_form=change_username_form,
@@ -605,6 +615,36 @@ def update_smtp_settings() -> Response | str:
         flash("👍 SMTP settings updated successfully")
         return redirect(url_for(".index"))
 
+    @bp.route("/update-brand-primary-color", methods=["POST"])
+    @admin_authentication_required
+    def update_brand_primary_color() -> Response | str:
+        host_org = HostOrganization.fetch_or_default()
+        form = UpdateBrandPrimaryColorForm()
+        if form.validate_on_submit():
+            host_org.brand_primary_hex_color = form.brand_primary_hex_color.data
+            db.session.add(host_org)  # explicitly add because instance might be new
+            db.session.commit()
+            flash("👍 Brand primary color updated successfully.")
+            return redirect(url_for(".index"))
+
+        flash("⛔ Invalid form data. Please try again.")
+        return redirect(url_for(".index"))
+
+    @bp.route("/update-brand-app-name", methods=["POST"])
+    @admin_authentication_required
+    def update_brand_app_name() -> Response | str:
+        host_org = HostOrganization.fetch_or_default()
+        form = UpdateBrandAppNameForm()
+        if form.validate_on_submit():
+            host_org.brand_app_name = form.brand_app_name.data
+            db.session.add(host_org)  # explicitly add because instance might be new
+            db.session.commit()
+            flash("👍 Brand app name updated successfully.")
+            return redirect(url_for(".index"))
+
+        flash("⛔ Invalid form data. Please try again.")
+        return redirect(url_for(".index"))
+
     @bp.route("/delete-account", methods=["POST"])
     @authentication_required
     def delete_account() -> Response | str:

hushline/settings/forms.py

@@ -15,7 +15,7 @@
 from wtforms.validators import DataRequired, Email, Length
 from wtforms.validators import Optional as OptionalField
 
-from ..forms import ComplexPassword
+from ..forms import CanonicalHTML, ComplexPassword, HexColor
 from ..model import SMTPEncryption
 
 
@@ -173,3 +173,13 @@ class ProfileForm(FlaskForm):
         filters=[strip_whitespace],
         validators=[OptionalField(), Length(max=4096)],
     )
+
+
+class UpdateBrandPrimaryColorForm(FlaskForm):
+    brand_primary_hex_color = StringField("Hex Color", validators=[DataRequired(), HexColor()])
+
+
+class UpdateBrandAppNameForm(FlaskForm):
+    brand_app_name = StringField(
+        "App Name", validators=[CanonicalHTML(), DataRequired(), Length(min=2, max=30)]
+    )