Merge pull request #588 from scidsg/username-aliases

Username aliases

.prettierignore

@@ -2,4 +2,5 @@
 build
 coverage
 hushline/static/vendor/*
+hushline/templates/*
 .pytest_cache

Makefile

@@ -13,10 +13,20 @@ install:
 
 
 .PHONY: run
-run: migrate ## Run the app
+run: ## Run the app
 	. ./dev_env.sh && \
 	poetry run flask run --debug -h localhost -p 8080
 
+.PHONY: migrate-dev
+migrate-dev: ## Run dev env migrations
+	. ./dev_env.sh && \
+	poetry run ./scripts/dev_migrations.py
+
+.PHONY: migrate-prod
+migrate-prod: ## Run prod env (alembic) migrations
+	. ./dev_env.sh && \
+	poetry run flask db upgrade
+
 .PHONY: lint
 lint: ## Lint the code
 	poetry run ruff format --check && \
@@ -30,13 +40,8 @@ fix: ## Format the code
 	poetry run ruff check --fix
 	npx prettier --write .
 
-.PHONY: migrate
-migrate: ## Apply migrations
-	. ./dev_env.sh && \
-	poetry run flask db upgrade
-
 .PHONY: revision
-revision:  ## Create a new migration
+revision: migrate-prod ## Create a new migration
 ifndef message
 	$(error 'message' must be set when invoking the revision target, eg `make revision message="short message"`)
 endif

docs/3-managed-service.md

@@ -210,8 +210,8 @@ The account's primary inbox will aggregate and label messages in a single view.
 
 <img src="img/paid.primary.inbox.png">
 
-#### Secondary Inboxes
+#### Aliased Inboxes
 
-Account owners can navigate to their secondary username's inboxes from the settings page.
+Account owners can navigate to their aliases' inboxes from the settings page.
 
-<img src="img/paid.secondary.inbox.png">
+<img src="img/paid.alias.inbox.png">

docs/DEV.md

@@ -41,10 +41,11 @@ Install Poetry dependencies:
 make install
 ```
 
-Run the database migrations:
+Run one of the database migrations:
 
 ```sh
-make migrate
+make migrate-dev  # for current dev DB settings
+make migrate-prod # for current alembic migrations
 ```
 
 Run the app in debug mode:

hushline/admin.py

@@ -15,7 +15,7 @@ def toggle_verified(user_id: int) -> Response:
         user = db.session.get(User, user_id)
         if user is None:
             abort(404)
-        user.is_verified = not user.is_verified
+        user.primary_username.is_verified = not user.primary_username.is_verified
         db.session.commit()
         flash("✅ User verification status toggled.", "success")
         return redirect(url_for("settings.index"))

hushline/make_admin.py

@@ -4,37 +4,25 @@
 
 from hushline import create_app
 from hushline.db import db
-from hushline.model import SecondaryUsername, User
+from hushline.models import Username
 
 
 def toggle_admin(username: str) -> None:
-    # First, try to find a primary user
-    user = db.session.scalars(db.select(User).filter_by(primary_username=username).limit(1)).first()
-
-    # If not found, try to find a secondary user
-    if not user:
-        secondary_username = db.session.scalars(
-            db.select(SecondaryUsername).filter_by(username=username).limit(1)
-        ).first()
-        if secondary_username:
-            user = secondary_username.primary_user
-        else:
-            print("User not found.")
-            return
-
-    # Toggle admin status
-    user.is_admin = not user.is_admin
+    uname = db.session.scalars(db.select(Username).filter_by(_username=username)).one_or_none()
+    if not uname:
+        print("User not found.")
+        return
+
+    uname.user.is_admin = not uname.user.is_admin
     db.session.commit()
 
-    print(f"User {username} admin status toggled to {user.is_admin}.")
+    print(f"User {username} admin status toggled to {uname.user.is_admin}.")
 
 
 if __name__ == "__main__":
     if len(sys.argv) != 2:  # noqa: PLR2004
         print("Usage: python make_admin.py <username>")
         sys.exit(1)
 
-    username = sys.argv[1]
-
     with create_app().app_context():
-        toggle_admin(username)
+        toggle_admin(sys.argv[1])

hushline/model.py

@@ -1,9 +1,9 @@
 import enum
 import secrets
+from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
-from typing import TYPE_CHECKING, Optional, Set
+from typing import TYPE_CHECKING, Any, Generator, Optional, Sequence
 
-from flask import current_app
 from flask_sqlalchemy.model import Model
 from passlib.hash import scrypt
 from sqlalchemy import Index
@@ -19,6 +19,7 @@
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 
+@enum.unique
 class SMTPEncryption(enum.Enum):
     SSL = "SSL"
     StartTLS = "StartTLS"
@@ -28,32 +29,32 @@ def default(cls) -> "SMTPEncryption":
         return cls.StartTLS
 
 
-class User(Model):
-    __tablename__ = "users"
+@dataclass(frozen=True, repr=False, eq=False)
+class ExtraField:
+    label: Optional[str]
+    value: Optional[str]
+    is_verified: Optional[bool]
+
+
+class Username(Model):
+    """
+    Class representing a username and associated profile.
+    This was pulled out of the `User` class so that a `username` could be globally unique among
+    both users and aliases and enforced at the database level.
+    """
+
+    __tablename__ = "usernames"
 
     id: Mapped[int] = mapped_column(primary_key=True)
-    primary_username: Mapped[str] = mapped_column(db.String(80), unique=True)
-    display_name: Mapped[Optional[str]] = mapped_column(db.String(80))
-    _password_hash: Mapped[str] = mapped_column("password_hash", db.String(512))
-    _totp_secret: Mapped[Optional[str]] = mapped_column("totp_secret", db.String(255))
-    _email: Mapped[Optional[str]] = mapped_column("email", db.String(255))
-    _smtp_server: Mapped[Optional[str]] = mapped_column("smtp_server", db.String(255))
-    smtp_port: Mapped[Optional[int]]
-    _smtp_username: Mapped[Optional[str]] = mapped_column("smtp_username", db.String(255))
-    _smtp_password: Mapped[Optional[str]] = mapped_column("smtp_password", db.String(255))
-    _pgp_key: Mapped[Optional[str]] = mapped_column("pgp_key", db.Text)
+    user_id: Mapped[int] = mapped_column(db.ForeignKey("users.id"))
+    user: Mapped["User"] = relationship()
+    _username: Mapped[str] = mapped_column("username", unique=True)
+    _display_name: Mapped[Optional[str]] = mapped_column("display_name", db.String(80))
+    is_primary: Mapped[bool] = mapped_column()
     is_verified: Mapped[bool] = mapped_column(default=False)
-    is_admin: Mapped[bool] = mapped_column(default=False)
     show_in_directory: Mapped[bool] = mapped_column(default=False)
     bio: Mapped[Optional[str]] = mapped_column(db.Text)
-    # Corrected the relationship and backref here
-    secondary_usernames: Mapped[Set["SecondaryUsername"]] = relationship(
-        backref=db.backref("primary_user", lazy=True)
-    )
-    smtp_encryption: Mapped[SMTPEncryption] = mapped_column(
-        db.Enum(SMTPEncryption, native_enum=False), default=SMTPEncryption.StartTLS
-    )
-    smtp_sender: Mapped[Optional[str]]
+
     extra_field_label1: Mapped[Optional[str]]
     extra_field_value1: Mapped[Optional[str]]
     extra_field_label2: Mapped[Optional[str]]
@@ -67,6 +68,84 @@ class User(Model):
     extra_field_verified3: Mapped[Optional[bool]] = mapped_column(default=False)
     extra_field_verified4: Mapped[Optional[bool]] = mapped_column(default=False)
 
+    def __init__(
+        self,
+        _username: str,
+        is_primary: bool,
+        **kwargs: Any,
+    ) -> None:
+        super().__init__(**kwargs)
+        self._username = _username
+        self.is_primary = is_primary
+
+    @property
+    def username(self) -> str:
+        return self._username
+
+    @username.setter
+    def username(self, username: str) -> None:
+        self._username = username
+        self.is_verified = False
+        db.session.commit()
+
+    @property
+    def display_name(self) -> Optional[str]:
+        return self._display_name
+
+    @display_name.setter
+    def display_name(self, display_name: str | None) -> None:
+        self._display_name = display_name
+        self.is_verified = False
+        db.session.commit()
+
+    @property
+    def extra_fields(self) -> Generator[ExtraField, None, None]:
+        for i in range(1, 5):
+            yield ExtraField(
+                getattr(self, f"extra_field_label{i}", None),
+                getattr(self, f"extra_field_value{i}", None),
+                getattr(self, f"extra_field_verified{i}", None),
+            )
+
+    @property
+    def valid_fields(self) -> Sequence[ExtraField]:
+        return [x for x in self.extra_fields if x.label and x.value]
+
+
+class User(Model):
+    __tablename__ = "users"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    is_admin: Mapped[bool] = mapped_column(default=False)
+    _password_hash: Mapped[str] = mapped_column("password_hash", db.String(512))
+    _totp_secret: Mapped[Optional[str]] = mapped_column("totp_secret", db.String(255))
+
+    primary_username: Mapped[Username] = relationship(
+        primaryjoin="and_(Username.user_id == User.id, Username.is_primary)",
+        back_populates="user",
+    )
+    messages: Mapped[list["Message"]] = relationship(
+        secondary="usernames",
+        primaryjoin="Username.user_id == User.id",
+        secondaryjoin="Message.username_id == Username.id",
+        order_by="Message.id.desc()",
+        backref=db.backref("user", lazy=False, uselist=False, viewonly=True),
+        lazy=True,
+        uselist=True,
+        viewonly=True,
+    )
+
+    _email: Mapped[Optional[str]] = mapped_column("email", db.String(255))
+    _smtp_server: Mapped[Optional[str]] = mapped_column("smtp_server", db.String(255))
+    smtp_port: Mapped[Optional[int]]
+    _smtp_username: Mapped[Optional[str]] = mapped_column("smtp_username", db.String(255))
+    _smtp_password: Mapped[Optional[str]] = mapped_column("smtp_password", db.String(255))
+    _pgp_key: Mapped[Optional[str]] = mapped_column("pgp_key", db.Text)
+    smtp_encryption: Mapped[SMTPEncryption] = mapped_column(
+        db.Enum(SMTPEncryption, native_enum=False), default=SMTPEncryption.StartTLS
+    )
+    smtp_sender: Mapped[Optional[str]]
+
     @property
     def password_hash(self) -> str:
         """Return the hashed password."""
@@ -135,40 +214,13 @@ def pgp_key(self, value: str) -> None:
         else:
             self._pgp_key = encrypt_field(value)
 
-    def update_display_name(self, new_display_name: str) -> None:
-        """Update the user's display name and remove verification status if the user is verified."""
-        self.display_name = new_display_name
-        if self.is_verified:
-            self.is_verified = False
-
-    # In the User model
-    def update_username(self, new_username: str) -> None:
-        """Update the user's username and remove verification status if the user is verified."""
-        try:
-            # Log the attempt to update the username
-            current_app.logger.debug(
-                f"Attempting to update username from {self.primary_username} to {new_username}"
-            )
-
-            # Update the username
-            self.primary_username = new_username
-            if self.is_verified:
-                self.is_verified = False
-                # Log the change in verification status due to username update
-                current_app.logger.debug("Verification status set to False due to username update")
-
-            # Commit the change to the database
-            db.session.commit()
-
-            # Log the successful update
-            current_app.logger.debug(f"Username successfully updated to {new_username}")
-        except Exception as e:
-            # Log any exceptions that occur during the update
-            current_app.logger.error(f"Error updating username: {e}", exc_info=True)
-
-    def __init__(self, primary_username: str) -> None:
+    def __init__(self, **kwargs: Any) -> None:
+        for key in ["password_hash", "_password_hash"]:
+            if key in kwargs:
+                raise ValueError(f"Key {key!r} cannot be mannually set. Try 'password' instead.")
+        pw = kwargs.pop("password", None)
         super().__init__()
-        self.primary_username = primary_username
+        self.password_hash = pw
 
 
 class AuthenticationLog(Model):
@@ -210,32 +262,17 @@ def __init__(
         self.timecode = timecode
 
 
-class SecondaryUsername(Model):
-    __tablename__ = "secondary_usernames"
-
-    id: Mapped[int] = mapped_column(primary_key=True)
-    username: Mapped[str] = mapped_column(db.String(80), unique=True)
-    # This foreign key points to the 'user' table's 'id' field
-    user_id: Mapped[int] = mapped_column(db.ForeignKey("users.id"))
-    display_name: Mapped[Optional[str]] = mapped_column(db.String(80))
-
-
 class Message(Model):
     id: Mapped[int] = mapped_column(primary_key=True)
     _content: Mapped[str] = mapped_column("content", db.Text)  # Encrypted content stored here
-    user_id: Mapped[int] = mapped_column(db.ForeignKey("users.id"))
-    user: Mapped["User"] = relationship(backref=db.backref("messages", lazy=True))
-    secondary_user_id: Mapped[Optional[int]] = mapped_column(
-        db.ForeignKey("secondary_usernames.id")
-    )
-    secondary_username: Mapped[Set["SecondaryUsername"]] = relationship(
-        "SecondaryUsername", backref="messages"
-    )
+    username_id: Mapped[int] = mapped_column(db.ForeignKey("usernames.id"))
+    username: Mapped["Username"] = relationship(uselist=False)
 
-    def __init__(self, content: str, user_id: int) -> None:
-        super().__init__()
+    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
-        self.user_id = user_id
 
     @property
     def content(self) -> str | None: