IP address plus cookie-based ban system

app/auth/routes.py

@@ -1,5 +1,5 @@
 from datetime import date, datetime, timedelta
-from flask import redirect, url_for, flash, request, make_response, session, Markup, current_app
+from flask import redirect, url_for, flash, request, make_response, session, Markup, current_app, g
 from werkzeug.urls import url_parse
 from flask_login import login_user, logout_user, current_user
 from flask_babel import _
@@ -7,10 +7,10 @@ from app import db
 from app.auth import bp
 from app.auth.forms import LoginForm, RegistrationForm, ResetPasswordRequestForm, ResetPasswordForm
 from app.auth.util import random_token
-from app.models import User, utcnow
+from app.models import User, utcnow, IpBan
 from app.auth.email import send_password_reset_email, send_welcome_email, send_verification_email
 from app.activitypub.signature import RsaKeys
-from app.utils import render_template
+from app.utils import render_template, ip_address, user_ip_banned, user_cookie_banned
 
 
 @bp.route('/login', methods=['GET', 'POST'])
@@ -29,9 +29,6 @@ def login():
         if user.deleted:
             flash(_('No account exists with that user name.'), 'error')
             return redirect(url_for('auth.login'))
-        if user.banned:
-            flash(_('You have been banned.'), 'error')
-            return redirect(url_for('auth.login'))
         if not user.check_password(form.password.data):
             if user.password_hash is None:
                 if "@gmail.com" in user.email:
@@ -43,9 +40,24 @@ def login():
                 return redirect(url_for('auth.login'))
             flash(_('Invalid password'))
             return redirect(url_for('auth.login'))
+        if user.banned or user_ip_banned() or user_cookie_banned():
+            flash(_('You have been banned.'), 'error')
+
+            response = make_response(redirect(url_for('auth.login')))
+            # Detect if a banned user tried to log in from a new IP address
+            if user.banned and not user_ip_banned():
+                # If so, ban their new IP address as well
+                new_ip_ban = IpBan(ip_address=ip_address(), notes=user.user_name + ' used new IP address')
+                db.session.add(new_ip_ban)
+                db.session.commit()
+
+            # Set a cookie so we have another way to track banned people
+            response.set_cookie('sesion', '17489047567495', expires=datetime(year=2099, month=12, day=30))
+            return response
         login_user(user, remember=True)
         current_user.last_seen = utcnow()
         current_user.verification_token = ''
+        current_user.ip_address = ip_address()
         db.session.commit()
         next_page = request.args.get('next')
         if not next_page or url_parse(next_page).netloc != '':
@@ -76,7 +88,8 @@ def register():
                 verification_token = random_token(16)
                 form.user_name.data = form.user_name.data.strip()
                 user = User(user_name=form.user_name.data, email=form.real_email.data,
-                            verification_token=verification_token, instance=1)
+                            verification_token=verification_token, instance=1, ipaddress=ip_address(),
+                            banned=user_ip_banned() or user_cookie_banned())
                 user.set_password(form.password.data)
                 db.session.add(user)
                 db.session.commit()
@@ -89,13 +102,11 @@ def register():
 
                 flash(_('Great, you are now a registered user!'))
 
-        # set a cookie so the login button is emphasised on the public site, for future visits
         resp = make_response(redirect(url_for('main.index')))
-        resp.set_cookie('logged_in_before', value='1', expires=datetime.now() + timedelta(weeks=300),
+        if user_ip_banned():
-                        domain='.chorebuster.net')
+            resp.set_cookie('sesion', '17489047567495', expires=datetime(year=2099, month=12, day=30))
         return resp
-    return render_template('auth/register.html', title=_('Register'),
+    return render_template('auth/register.html', title=_('Register'), form=form, site=g.site)
-                           form=form)
 
 
 @bp.route('/reset_password_request', methods=['GET', 'POST'])

app/models.py

@@ -260,6 +260,7 @@ class User(UserMixin, db.Model):
     bot = db.Column(db.Boolean, default=False)
     ignore_bots = db.Column(db.Boolean, default=False)
     unread_notifications = db.Column(db.Integer, default=0)
+    ip_address = db.Column(db.String(50))
     instance_id = db.Column(db.Integer, db.ForeignKey('instance.id'), index=True)
 
     avatar = db.relationship('File', lazy='joined', foreign_keys=[avatar_id], single_parent=True, cascade="all, delete-orphan")
@@ -870,6 +871,13 @@ class Report(db.Model):
     updated = db.Column(db.DateTime, default=utcnow)
 
 
+class IpBan(db.Model):
+    id = db.Column(db.Integer, primary_key=True)
+    ip_address = db.Column(db.String(50), index=True)
+    notes = db.Column(db.String(150))
+    created_at = db.Column(db.DateTime, default=utcnow)
+
+
 class Site(db.Model):
     id = db.Column(db.Integer, primary_key=True)
     name = db.Column(db.String(256))

app/post/routes.py

@@ -18,7 +18,7 @@ from app.models import Post, PostReply, \
 from app.post import bp
 from app.utils import get_setting, render_template, allowlist_html, markdown_to_html, validation_required, \
     shorten_string, markdown_to_text, domain_from_url, validate_image, gibberish, ap_datetime, return_304, \
-    request_etag_matches
+    request_etag_matches, ip_address
 
 
 def show_post(post_id: int):
@@ -224,6 +224,7 @@ def post_vote(post_id: int, vote_direction):
                 flash('Failed to send vote', 'warning')
 
     current_user.last_seen = utcnow()
+    current_user.ip_address = ip_address()
     db.session.commit()
     current_user.recalculate_attitude()
     db.session.commit()
@@ -296,6 +297,7 @@ def comment_vote(comment_id, vote_direction):
                 flash('Failed to send vote', 'error')
 
     current_user.last_seen = utcnow()
+    current_user.ip_address = ip_address()
     db.session.commit()
     current_user.recalculate_attitude()
     db.session.commit()
@@ -333,6 +335,7 @@ def add_reply(post_id: int, comment_id: int):
     form = NewReplyForm()
     if form.validate_on_submit():
         current_user.last_seen = utcnow()
+        current_user.ip_address = ip_address()
         reply = PostReply(user_id=current_user.id, post_id=post.id, parent_id=in_reply_to.id, depth=in_reply_to.depth + 1,
                           community_id=post.community.id, body=form.body.data,
                           body_html=markdown_to_html(form.body.data), body_html_safe=True,

app/templates/auth/register.html

@@ -7,8 +7,12 @@
             <div class="col col-login mx-auto">
                 <div class="card mt-5">
                     <div class="card-body p-6" id="registration_form">
-                        <div class="card-title text-center">{{ _('Create new account') }}</div>
+                        {% if site.registration_mode != 'Closed' %}
-                        {{ render_form(form) }}
+                            <div class="card-title text-center">{{ _('Create new account') }}</div>
+                            {{ render_form(form) }}
+                        {% else %}
+                            {{ _('Registration is closed. Only admins can create accounts.') }}
+                        {% endif %}
                     </div>
                 </div>
             </div>

app/utils.py

@@ -21,7 +21,7 @@ from wtforms.fields  import SelectField, SelectMultipleField
 from wtforms.widgets import Select, html_params, ListWidget, CheckboxInput
 from app import db, cache
 
-from app.models import Settings, Domain, Instance, BannedInstances, User, Community, DomainBlock, ActivityPubLog
+from app.models import Settings, Domain, Instance, BannedInstances, User, Community, DomainBlock, ActivityPubLog, IpBan
 
 
 # Flask's render_template function, with support for themes added
@@ -344,4 +344,28 @@ def ap_datetime(date_time: datetime) -> str:
 
 class MultiCheckboxField(SelectMultipleField):
     widget = ListWidget(prefix_label=False)
-    option_widget = CheckboxInput()
+    option_widget = CheckboxInput()
+
+
+def ip_address() -> str:
+    ip = request.headers.get('X-Forwarded-For') or request.remote_addr
+    if ',' in ip:  # Remove all but first ip addresses
+        ip = ip[:ip.index(',')].strip()
+    return ip
+
+
+def user_ip_banned() -> bool:
+    current_ip_address = ip_address()
+    if current_ip_address:
+        return current_ip_address in banned_ip_addresses()
+
+
+def user_cookie_banned() -> bool:
+    cookie = request.cookies.get('sesion', None)
+    return cookie is not None
+
+
+@cache.cached(timeout=300)
+def banned_ip_addresses() -> List[str]:
+    ips = IpBan.query.all()
+    return [ip.ip_address for ip in ips]

migrations/versions/3f17b9ab55e4_ip_ban.py

@@ -0,0 +1,46 @@
+"""ip ban
+
+Revision ID: 3f17b9ab55e4
+Revises: b58c4301d1ad
+Create Date: 2023-12-30 17:23:32.363913
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '3f17b9ab55e4'
+down_revision = 'b58c4301d1ad'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('ip_ban',
+    sa.Column('id', sa.Integer(), nullable=False),
+    sa.Column('ip_address', sa.String(length=50), nullable=True),
+    sa.Column('notes', sa.String(length=150), nullable=True),
+    sa.Column('created_at', sa.DateTime(), nullable=True),
+    sa.PrimaryKeyConstraint('id')
+    )
+    with op.batch_alter_table('ip_ban', schema=None) as batch_op:
+        batch_op.create_index(batch_op.f('ix_ip_ban_ip_address'), ['ip_address'], unique=False)
+
+    with op.batch_alter_table('user', schema=None) as batch_op:
+        batch_op.add_column(sa.Column('ip_address', sa.String(length=50), nullable=True))
+
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('user', schema=None) as batch_op:
+        batch_op.drop_column('ip_address')
+
+    with op.batch_alter_table('ip_ban', schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f('ix_ip_ban_ip_address'))
+
+    op.drop_table('ip_ban')
+    # ### end Alembic commands ###