Add User Status Checks on Endpoints

ambuda/auth.py

@@ -19,7 +19,8 @@ def _load_user(user_id: int) -> Optional[User]:
     import current_user`) and as a template variable injected into each template.
     """
     session = get_session()
-    return session.query(User).get(int(user_id))
+    user = session.query(User).get(int(user_id))
+    return user if user and user.is_active else None
 
 
 def _unauthorized():

ambuda/checks.py

@@ -134,7 +134,11 @@ def _check_lookup_tables(session) -> list[str]:
 def _check_bot_user(session) -> list[str]:
     """Check that the ambuda-bot user exists."""
     username = consts.BOT_USERNAME
-    bot_user = session.query(db.User).filter_by(username=username).first()
+    bot_user = (
+        session.query(db.User)
+        .filter_by(username=username, status=enums.UserStatus.ACTIVE.value)
+        .first()
+    )
     if bot_user:
         return []
     else:

ambuda/enums.py

@@ -27,3 +27,17 @@ class SitePageStatus(str, Enum):
     R2 = "reviewed-2"
     #: Not relevant.
     SKIP = "skip"
+
+
+class UserStatus(str, Enum):
+    """Defines status of user on Ambuda."""
+
+    #: Active user whose activity is visible to others.
+    ACTIVE = "active"
+    #: User that deleted their account. Their created projects are still visible
+    #: (unless explicitly deleted) however their discussions posts and
+    #: threads are removed. Their username is replaced with "deleted_user".
+    DELETED = "deleted"
+    #: User was banned by someone with moderator permissions.
+    #: Visiblity is the same as `DELETED`.
+    BANNED = "banned"

ambuda/models/auth.py

@@ -7,6 +7,7 @@
 
 from ambuda.models.base import Base, foreign_key, pk
 from ambuda.utils.user_mixins import AmbudaUserMixin
+from ambuda.enums import UserStatus
 
 
 class User(AmbudaUserMixin, Base):
@@ -28,13 +29,20 @@ class User(AmbudaUserMixin, Base):
     #: The user's self-description.
     description = Column(Text_, nullable=False, default="")
 
+    #: The user's status.
+    status = Column(String, nullable=False, default=UserStatus.ACTIVE.value)
+
     #: All roles available for this user.
     roles = relationship("Role", secondary="user_roles")
 
     def set_password(self, raw_password: str):
         """Hash and save the given password."""
         self.password_hash = generate_password_hash(raw_password)
 
+    def set_status(self, status: UserStatus):
+        """Update user status."""
+        self.status = status.value
+
     def check_password(self, raw_password: str) -> bool:
         """Check if the given password matches the user's hash."""
         return check_password_hash(self.password_hash, raw_password)

ambuda/queries.py

@@ -12,6 +12,7 @@
 from sqlalchemy.orm import load_only, scoped_session, selectinload, sessionmaker
 
 import ambuda.database as db
+from ambuda.enums import UserStatus
 
 # NOTE: this logic is copied from Flask-SQLAlchemy. We avoid Flask-SQLAlchemy
 # because we also need to access the database from a non-Flask context when
@@ -204,7 +205,11 @@ def page(project_id, page_slug: str) -> Optional[db.Page]:
 
 def user(username: str) -> Optional[db.User]:
     session = get_session()
-    return session.query(db.User).filter_by(username=username).first()
+    return (
+        session.query(db.User)
+        .filter_by(username=username, status=UserStatus.ACTIVE.value)
+        .first()
+    )
 
 
 def create_user(*, username: str, email: str, raw_password: str) -> db.User:

ambuda/seed/lookup/create_bot_user.py

@@ -25,7 +25,9 @@ def run():
     engine = create_db()
     logging.debug("Creating bot user ...")
     with Session(engine) as session:
-        user = session.query(db.User).filter_by(username=consts.BOT_USERNAME).first()
+        user = (
+            session.query(db.User).filter_by(username=consts.BOT_USERNAME).first()
+        )
         if not user:
             _create_bot_user(session)
     logging.debug("Done.")

ambuda/utils/user_mixins.py

@@ -2,7 +2,7 @@
 
 from flask_login import AnonymousUserMixin, UserMixin
 
-from ambuda.enums import SiteRole
+from ambuda.enums import SiteRole, UserStatus
 
 
 class AmbudaAnonymousUser(AnonymousUserMixin):
@@ -31,6 +31,10 @@ def is_moderator(self) -> bool:
     def is_admin(self) -> bool:
         return False
 
+    @property
+    def is_active(self) -> bool:
+        return True
+
 
 class AmbudaUserMixin(UserMixin):
     def has_role(self, role: SiteRole) -> bool:
@@ -59,3 +63,7 @@ def is_moderator(self) -> bool:
     @property
     def is_admin(self) -> bool:
         return self.has_role(SiteRole.ADMIN)
+
+    @property
+    def is_active(self) -> bool:
+        return self.status == UserStatus.ACTIVE.value

ambuda/views/auth.py

@@ -129,6 +129,7 @@ class ResetPasswordFromTokenForm(FlaskForm):
 @bp.route("/register", methods=["GET", "POST"])
 def register():
     if current_user.is_authenticated:
+        handle_user_status_change()
         return redirect(url_for("site.index"))
 
     form = SignupForm()
@@ -152,6 +153,7 @@ def register():
 @bp.route("/sign-in", methods=["GET", "POST"])
 def sign_in():
     if current_user.is_authenticated:
+        handle_user_status_change()
         return redirect(url_for("site.index"))
 
     form = SignInForm()
@@ -165,6 +167,13 @@ def sign_in():
     return render_template("auth/sign-in.html", form=form)
 
 
+def handle_user_status_change():
+    # Check if user is now delted or banned
+    user = q.user(username=current_user.username)
+    if user and not user.is_active:
+        logout_user()
+
+
 @bp.route("/sign-out")
 def sign_out():
     logout_user()

migrations/versions/99379162619e_add_user_status.py

@@ -0,0 +1,46 @@
+"""Add user status
+
+Revision ID: 99379162619e
+Revises: bc48af5ec2e6
+Create Date: 2022-09-11 17:20:02.341713
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy import orm
+
+from ambuda.enums import UserStatus
+
+# revision identifiers, used by Alembic.
+revision = "99379162619e"
+down_revision = "bc48af5ec2e6"
+branch_labels = None
+depends_on = None
+
+Base = orm.declarative_base()
+
+
+class User(Base):
+    __tablename__ = "users"
+    id = sa.Column(sa.Integer, primary_key=True)
+    status = sa.Column(sa.String, default=UserStatus.ACTIVE.value, nullable=False)
+
+
+def upgrade() -> None:
+    op.add_column("users", sa.Column("status", sa.String, nullable=True))
+
+    bind = op.get_bind()
+    session = orm.Session(bind=bind)
+    for user in session.query(User).all():
+        user.status = UserStatus.ACTIVE.value
+        session.add(user)
+    session.commit()
+
+    with op.batch_alter_table("users") as batch_op:
+        batch_op.alter_column("status", existing_type=sa.VARCHAR(), nullable=False)
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column("users", "status")
+    # ### end Alembic commands ###

test/ambuda/conftest.py

@@ -4,6 +4,7 @@
 import ambuda.database as db
 from ambuda import create_app
 from ambuda.consts import BOT_USERNAME, TEXT_CATEGORIES
+from ambuda.enums import UserStatus
 from ambuda.queries import get_engine, get_session
 
 
@@ -73,6 +74,19 @@ def initialize_test_db():
     session.add(admin)
     session.flush()
 
+    # Deleted and Banned
+    deleted_admin = db.User(username="sandrocottus", email="cgm@ambuda.org")
+    deleted_admin.set_password("maurya")
+    deleted_admin.set_status(UserStatus.DELETED)
+
+    banned = db.User(username="sikander", email="alex@ambuda.org")
+    banned.set_password("onesicritus")
+    banned.set_status(UserStatus.BANNED)
+
+    session.add(deleted_admin)
+    session.add(banned)
+    session.flush()
+
     # Roles
     p1_role = db.Role(name=db.SiteRole.P1.value)
     p2_role = db.Role(name=db.SiteRole.P2.value)
@@ -84,8 +98,12 @@ def initialize_test_db():
 
     rama.roles = [p1_role, p2_role]
     admin.roles = [p1_role, p2_role, admin_role]
+    deleted_admin.roles = [p1_role, p2_role, admin_role]
+    banned.roles = [p1_role]
     session.add(rama)
     session.add(admin)
+    session.add(deleted_admin)
+    session.add(banned)
     session.flush()
 
     # Proofreading
@@ -152,3 +170,17 @@ def admin_client(flask_app):
     session = get_session()
     user = session.query(db.User).filter_by(username="akprasad").first()
     return flask_app.test_client(user=user)
+
+
+@pytest.fixture()
+def deleted_client(flask_app):
+    session = get_session()
+    user = session.query(db.User).filter_by(username="sandrocottus").first()
+    return flask_app.test_client(user=user)
+
+
+@pytest.fixture()
+def banned_client(flask_app):
+    session = get_session()
+    user = session.query(db.User).filter_by(username="sikander").first()
+    return flask_app.test_client(user=user)

test/ambuda/test_models.py

@@ -1,4 +1,5 @@
 import ambuda.database as db
+from ambuda.enums import UserStatus
 from ambuda.queries import get_session
 
 
@@ -38,6 +39,25 @@ def test_user__set_and_check_role(client):
     _cleanup(session, user)
 
 
+def test_user__deletion(client):
+    session = get_session()
+
+    # Check active user
+    user = db.User(username="test", email="test@ambuda.org")
+    user.set_password("my-password")
+    session.add(user)
+    session.commit()
+    assert user.is_active
+
+    # Deleted
+    user.set_status(UserStatus.DELETED)
+    session.add(user)
+    session.commit()
+    assert not user.is_active
+
+    _cleanup(session, user)
+
+
 def test_role__repr(client):
     role = db.Role(name="foo")
     assert repr(role) == "<Role(None, 'foo')>"

test/ambuda/views/test_admin.py

@@ -8,6 +8,17 @@ def test_admin_index__auth(admin_client):
     assert resp.status_code == 200
 
 
+def test_admin_index__inactive(deleted_client, banned_client):
+    assert deleted_client.get("/admin/").status_code == 404
+    assert banned_client.get("/admin/").status_code == 404
+
+
 def test_admin_text__unauth(client):
     resp = client.get("/admin/text/")
     assert resp.status_code == 404
+
+
+def test_admin_text__inactive(deleted_client, banned_client):
+    assert deleted_client.get("/admin/text/").status_code == 404
+    assert banned_client.get("/admin/text/").status_code == 404
+

test/ambuda/views/test_auth.py

@@ -72,6 +72,13 @@ def test_register__auth(rama_client):
     assert r.status_code == 302
 
 
+def test_register__banned(banned_client):
+    with banned_client:
+        r = banned_client.get("/register")
+        assert r.status_code == 200
+        assert current_user.is_anonymous
+
+
 def test_sign_in__unauth(client):
     r = client.get("/sign-in")
     assert ">Sign in to Ambuda<" in r.text
@@ -115,6 +122,13 @@ def test_sign_in__auth(rama_client):
     assert r.status_code == 302
 
 
+def test_sign_in__banned(banned_client):
+    with banned_client:
+        r = banned_client.get("/sign-in")
+        assert r.status_code == 200
+        assert current_user.is_anonymous
+
+
 def test_sign_out__unauth(client):
     with client:
         r = client.get("/sign-out")