Linting and typing fixes; cleanups

securedrop/alembic/versions/c5a02eb52f2d_dropped_session_nonce_from_journalist_.py

@@ -5,88 +5,95 @@
 Create Date: 2022-04-16 21:25:22.398189
 
 """
-from alembic import op
 import sqlalchemy as sa
-
+from alembic import op
 
 # revision identifiers, used by Alembic.
-revision = 'c5a02eb52f2d'
-down_revision = 'b7f98cfd6a70'
+revision = "c5a02eb52f2d"
+down_revision = "b7f98cfd6a70"
 branch_labels = None
 depends_on = None
 
 
 def upgrade() -> None:
     # ### commands auto generated by Alembic - please adjust! ###
-    op.drop_table('revoked_tokens')
-    with op.batch_alter_table('journalists', schema=None) as batch_op:
-        batch_op.drop_column('session_nonce')
+    op.drop_table("revoked_tokens")
+    with op.batch_alter_table("journalists", schema=None) as batch_op:
+        batch_op.drop_column("session_nonce")
 
     # ### end Alembic commands ###
 
 
 def downgrade() -> None:
-    '''This would have been the easy way, however previous does not have
-       default value and thus up/down assertion fails'''
-    #op.add_column('journalists', sa.Column('session_nonce', sa.Integer(), nullable=False, server_default='0'))
+    """This would have been the easy way, however previous does not have
+    default value and thus up/down assertion fails"""
+    # op.add_column('journalists', sa.Column('session_nonce', sa.Integer(), nullable=False, server_default='0'))
 
     conn = op.get_bind()
     conn.execute("PRAGMA legacy_alter_table=ON")
     # Save existing journalist table.
-    op.rename_table('journalists', 'journalists_tmp')
+    op.rename_table("journalists", "journalists_tmp")
 
     # Add nonce column.
-    op.add_column('journalists_tmp', sa.Column('session_nonce', sa.Integer()))
+    op.add_column("journalists_tmp", sa.Column("session_nonce", sa.Integer()))
 
     # Populate nonce column.
-    journalists = conn.execute(
-        sa.text("SELECT * FROM journalists_tmp")).fetchall()
+    journalists = conn.execute(sa.text("SELECT * FROM journalists_tmp")).fetchall()
 
     for journalist in journalists:
         conn.execute(
-            sa.text("""UPDATE journalists_tmp SET session_nonce=0 WHERE
-                       id=:id""").bindparams(id=journalist.id)
-            )
+            sa.text(
+                """UPDATE journalists_tmp SET session_nonce=0 WHERE
+                       id=:id"""
+            ).bindparams(id=journalist.id)
+        )
 
     # Now create new table with null constraint applied.
-    op.create_table('journalists',
-        sa.Column('id', sa.Integer(), nullable=False),
-        sa.Column('uuid', sa.String(length=36), nullable=False),
-        sa.Column('username', sa.String(length=255), nullable=False),
-        sa.Column('first_name', sa.String(length=255), nullable=True),
-        sa.Column('last_name', sa.String(length=255), nullable=True),
-        sa.Column('pw_salt', sa.Binary(), nullable=True),
-        sa.Column('pw_hash', sa.Binary(), nullable=True),
-        sa.Column('passphrase_hash', sa.String(length=256), nullable=True),
-        sa.Column('is_admin', sa.Boolean(), nullable=True),
-        sa.Column('session_nonce', sa.Integer(), nullable=False),
-        sa.Column('otp_secret', sa.String(length=32), nullable=True),
-        sa.Column('is_totp', sa.Boolean(), nullable=True),
-        sa.Column('hotp_counter', sa.Integer(), nullable=True),
-        sa.Column('last_token', sa.String(length=6), nullable=True),
-        sa.Column('created_on', sa.DateTime(), nullable=True),
-        sa.Column('last_access', sa.DateTime(), nullable=True),
-        sa.PrimaryKeyConstraint('id'),
-        sa.UniqueConstraint('username'),
-        sa.UniqueConstraint('uuid')
+    op.create_table(
+        "journalists",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("uuid", sa.String(length=36), nullable=False),
+        sa.Column("username", sa.String(length=255), nullable=False),
+        sa.Column("first_name", sa.String(length=255), nullable=True),
+        sa.Column("last_name", sa.String(length=255), nullable=True),
+        sa.Column("pw_salt", sa.Binary(), nullable=True),
+        sa.Column("pw_hash", sa.Binary(), nullable=True),
+        sa.Column("passphrase_hash", sa.String(length=256), nullable=True),
+        sa.Column("is_admin", sa.Boolean(), nullable=True),
+        sa.Column("session_nonce", sa.Integer(), nullable=False),
+        sa.Column("otp_secret", sa.String(length=32), nullable=True),
+        sa.Column("is_totp", sa.Boolean(), nullable=True),
+        sa.Column("hotp_counter", sa.Integer(), nullable=True),
+        sa.Column("last_token", sa.String(length=6), nullable=True),
+        sa.Column("created_on", sa.DateTime(), nullable=True),
+        sa.Column("last_access", sa.DateTime(), nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("username"),
+        sa.UniqueConstraint("uuid"),
     )
 
-    conn.execute('''
+    conn.execute(
+        """
         INSERT INTO journalists
         SELECT id, uuid, username, first_name, last_name, pw_salt, pw_hash,
                passphrase_hash, is_admin, session_nonce, otp_secret, is_totp,
                hotp_counter, last_token, created_on, last_access
         FROM journalists_tmp
-    ''')
+    """
+    )
 
     # Now delete the old table.
-    op.drop_table('journalists_tmp')
-
-    op.create_table('revoked_tokens',
-                    sa.Column('id', sa.INTEGER(), nullable=False),
-                    sa.Column('journalist_id', sa.INTEGER(), nullable=False),
-                    sa.Column('token', sa.TEXT(), nullable=False),
-                    sa.ForeignKeyConstraint(['journalist_id'], ['journalists.id'], ),
-                    sa.PrimaryKeyConstraint('id'),
-                    sa.UniqueConstraint('token')
+    op.drop_table("journalists_tmp")
+
+    op.create_table(
+        "revoked_tokens",
+        sa.Column("id", sa.INTEGER(), nullable=False),
+        sa.Column("journalist_id", sa.INTEGER(), nullable=False),
+        sa.Column("token", sa.TEXT(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["journalist_id"],
+            ["journalists.id"],
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("token"),
     )

securedrop/journalist_app/__init__.py

@@ -1,35 +1,22 @@
 # -*- coding: utf-8 -*-
 
 import typing
-from datetime import datetime, timedelta, timezone
-from pathlib import Path
 from datetime import datetime
-from flask import (Flask, redirect, url_for, g, request,
-                   render_template, json, abort)
-from flask_assets import Environment
-from flask_babel import gettext
-from flask_wtf.csrf import CSRFProtect, CSRFError
 from os import path
 from pathlib import Path
 
 import i18n
 import template_filters
 import version
 from db import db
-from flask import Flask, flash, g, json, redirect, render_template, request, session, url_for
+from flask import Flask, abort, g, json, redirect, render_template, request, url_for
 from flask_babel import gettext
 from flask_wtf.csrf import CSRFError, CSRFProtect
 from journalist_app import account, admin, api, col, main
-from journalist_app.utils import (
-    JournalistInterfaceSessionInterface,
-    cleanup_expired_revoked_tokens,
-    get_source,
-    logged_in,
-)
-from journalist_app import account, admin, api, main, col
 from journalist_app.sessions import Session, session
 from journalist_app.utils import get_source
 from models import InstanceConfig
+from werkzeug.exceptions import default_exceptions
 
 # https://www.python.org/dev/peps/pep-0484/#runtime-or-type-checking
 if typing.TYPE_CHECKING:
@@ -43,8 +30,8 @@
     from werkzeug import Response  # noqa: F401
     from werkzeug.exceptions import HTTPException  # noqa: F401
 
-_insecure_views = ['main.login', 'static']
-_insecure_api_views = ['api.get_token', 'api.get_endpoints']
+_insecure_views = ["main.login", "static"]
+_insecure_api_views = ["api.get_token", "api.get_endpoints"]
 # Timezone-naive datetime format expected by SecureDrop Client
 API_DATETIME_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ"
 
@@ -96,9 +83,9 @@ def default(self, obj: "Any") -> "Any":
     @app.errorhandler(CSRFError)  # type: ignore
     def handle_csrf_error(e: CSRFError) -> "Response":
         app.logger.error("The CSRF token is invalid.")
-        msg = gettext('You have been logged out due to inactivity.')
-        session.destroy(('error', msg), session.get('locale'))
-        return redirect(url_for('main.login'))
+        msg = gettext("You have been logged out due to inactivity.")
+        session.destroy(("error", msg), session.get("locale"))
+        return redirect(url_for("main.login"))
 
     def _handle_http_exception(
         error: "HTTPException",
@@ -149,12 +136,12 @@ def setup_g() -> "Optional[Response]":
         except FileNotFoundError:
             app.logger.error("Site logo not found.")
 
-        if request.path.split('/')[1] == 'api':
+        if request.path.split("/")[1] == "api":
             if request.endpoint not in _insecure_api_views and not session.logged_in():
                 abort(403)
         else:
             if request.endpoint not in _insecure_views and not session.logged_in():
-                return redirect(url_for('main.login'))
+                return redirect(url_for("main.login"))
 
         if request.method == "POST":
             filesystem_id = request.form.get("filesystem_id")

securedrop/journalist_app/account.py

@@ -2,14 +2,16 @@
 from typing import Union
 
 import werkzeug
-from flask import (Blueprint, render_template, request, g, redirect, url_for,
-                   flash)
-from flask_babel import gettext
-
 from db import db
-from journalist_app.sessions import session
-from journalist_app.utils import (set_diceware_password, set_name, validate_user,
-                                  validate_hotp_secret)
+from flask import Blueprint, flash, g, redirect, render_template, request, url_for
+from flask_babel import gettext
+from journalist_app.sessions import logout_user, session
+from journalist_app.utils import (
+    set_diceware_password,
+    set_name,
+    validate_hotp_secret,
+    validate_user,
+)
 from passphrases import PassphraseGenerator
 from sdconfig import SDConfig
 
@@ -26,40 +28,42 @@ def edit() -> str:
 
     @view.route("/change-name", methods=("POST",))
     def change_name() -> werkzeug.Response:
-        first_name = request.form.get('first_name')
-        last_name = request.form.get('last_name')
+        first_name = request.form.get("first_name")
+        last_name = request.form.get("last_name")
         set_name(session.get_user(), first_name, last_name)
-        return redirect(url_for('account.edit'))
+        return redirect(url_for("account.edit"))
 
     @view.route("/new-password", methods=("POST",))
     def new_password() -> werkzeug.Response:
         user = session.get_user()
-        current_password = request.form.get('current_password')
-        token = request.form.get('token')
-        error_message = gettext('Incorrect password or two-factor code.')
+        current_password = request.form.get("current_password")
+        token = request.form.get("token")
+        error_message = gettext("Incorrect password or two-factor code.")
         # If the user is validated, change their password
-        if validate_user(user.username, current_password, token,
-                         error_message):
-            password = request.form.get('password')
+        if validate_user(user.username, current_password, token, error_message):
+            password = request.form.get("password")
             if set_diceware_password(user, password):
-                return redirect(url_for('main.login'))
-        return redirect(url_for('account.edit'))
+                logout_user(user.id)
+                return redirect(url_for("main.login"))
+        return redirect(url_for("account.edit"))
 
     @view.route("/2fa", methods=("GET", "POST"))
     def new_two_factor() -> Union[str, werkzeug.Response]:
-        if request.method == 'POST':
-            token = request.form['token']
+        if request.method == "POST":
+            token = request.form["token"]
             if session.get_user().verify_token(token):
-                flash(gettext("Your two-factor credentials have been reset successfully."),
-                      "notification")
-                return redirect(url_for('account.edit'))
+                flash(
+                    gettext("Your two-factor credentials have been reset successfully."),
+                    "notification",
+                )
+                return redirect(url_for("account.edit"))
             else:
                 flash(
                     gettext("There was a problem verifying the two-factor code. Please try again."),
                     "error",
                 )
 
-        return render_template('account_new_two_factor.html', user=session.get_user())
+        return render_template("account_new_two_factor.html", user=session.get_user())
 
     @view.route("/reset-2fa-totp", methods=["POST"])
     def reset_two_factor_totp() -> werkzeug.Response:
@@ -73,7 +77,7 @@ def reset_two_factor_hotp() -> Union[str, werkzeug.Response]:
         otp_secret = request.form.get("otp_secret", None)
         if otp_secret:
             if not validate_hotp_secret(session.get_user(), otp_secret):
-                return render_template('account_edit_hotp_secret.html')
+                return render_template("account_edit_hotp_secret.html")
             session.get_user().set_hotp_secret(otp_secret)
             db.session.commit()
             return redirect(url_for("account.new_two_factor"))