Add type annotations to journalist app

Fix types

Return HTTP 405 for wrong method

mypy.ini

@@ -3,13 +3,7 @@ ignore_missing_imports = True
 no_implicit_optional = True
 python_version = 3.5
 
-[mypy-journalist_app.utils]
-disallow_untyped_defs = True
-
-[mypy-journalist_app.models]
-disallow_untyped_defs = True
-
-[mypy-journalist_app.api]
+[mypy-journalist_app.*]
 disallow_untyped_defs = True
 
 [mypy-config]

securedrop/journalist_app/admin.py

@@ -1,7 +1,10 @@
 # -*- coding: utf-8 -*-
 
 import os
+from typing import Optional
+from typing import Union
 
+import werkzeug
 from flask import (Blueprint, render_template, request, url_for, redirect, g,
                    current_app, flash, abort)
 from flask_babel import gettext
@@ -15,26 +18,30 @@
 from journalist_app.utils import (make_password, commit_account_changes, set_diceware_password,
                                   validate_hotp_secret, revoke_token)
 from journalist_app.forms import LogoForm, NewUserForm, SubmissionPreferencesForm
+from sdconfig import SDConfig
 
 
-def make_blueprint(config):
+def make_blueprint(config: SDConfig) -> Blueprint:
     view = Blueprint('admin', __name__)
 
     @view.route('/', methods=('GET', 'POST'))
     @admin_required
-    def index():
+    def index() -> str:
         users = Journalist.query.all()
         return render_template("admin.html", users=users)
 
     @view.route('/config', methods=('GET', 'POST'))
     @admin_required
-    def manage_config():
+    def manage_config() -> Union[str, werkzeug.Response]:
         # The UI prompt ("prevent") is the opposite of the setting ("allow"):
         submission_preferences_form = SubmissionPreferencesForm(
             prevent_document_uploads=not current_app.instance_config.allow_document_uploads)
         logo_form = LogoForm()
         if logo_form.validate_on_submit():
             f = logo_form.logo.data
+
+            if current_app.static_folder is None:
+                abort(500)
             custom_logo_filepath = os.path.join(current_app.static_folder, 'i',
                                                 'custom_logo.png')
             try:
@@ -55,18 +62,20 @@ def manage_config():
 
     @view.route('/update-submission-preferences', methods=['POST'])
     @admin_required
-    def update_submission_preferences():
+    def update_submission_preferences() -> Optional[werkzeug.Response]:
         form = SubmissionPreferencesForm()
         if form.validate_on_submit():
             # The UI prompt ("prevent") is the opposite of the setting ("allow"):
             flash(gettext("Preferences saved."), "submission-preferences-success")
             value = not bool(request.form.get('prevent_document_uploads'))
             InstanceConfig.set('allow_document_uploads', value)
             return redirect(url_for('admin.manage_config'))
+        else:
+            return None
 
     @view.route('/add', methods=('GET', 'POST'))
     @admin_required
-    def add_user():
+    def add_user() -> Union[str, werkzeug.Response]:
         form = NewUserForm()
         if form.validate_on_submit():
             form_valid = True
@@ -121,7 +130,7 @@ def add_user():
 
     @view.route('/2fa', methods=('GET', 'POST'))
     @admin_required
-    def new_user_two_factor():
+    def new_user_two_factor() -> Union[str, werkzeug.Response]:
         user = Journalist.query.get(request.args['uid'])
 
         if request.method == 'POST':
@@ -141,7 +150,7 @@ def new_user_two_factor():
 
     @view.route('/reset-2fa-totp', methods=['POST'])
     @admin_required
-    def reset_two_factor_totp():
+    def reset_two_factor_totp() -> werkzeug.Response:
         uid = request.form['uid']
         user = Journalist.query.get(uid)
         user.is_totp = True
@@ -151,7 +160,7 @@ def reset_two_factor_totp():
 
     @view.route('/reset-2fa-hotp', methods=['POST'])
     @admin_required
-    def reset_two_factor_hotp():
+    def reset_two_factor_hotp() -> Union[str, werkzeug.Response]:
         uid = request.form['uid']
         otp_secret = request.form.get('otp_secret', None)
         if otp_secret:
@@ -165,7 +174,7 @@ def reset_two_factor_hotp():
 
     @view.route('/edit/<int:user_id>', methods=('GET', 'POST'))
     @admin_required
-    def edit_user(user_id):
+    def edit_user(user_id: int) -> Union[str, werkzeug.Response]:
         user = Journalist.query.get(user_id)
 
         if request.method == 'POST':
@@ -218,7 +227,7 @@ def edit_user(user_id):
 
     @view.route('/delete/<int:user_id>', methods=('POST',))
     @admin_required
-    def delete_user(user_id):
+    def delete_user(user_id: int) -> werkzeug.Response:
         user = Journalist.query.get(user_id)
         if user_id == g.user.id:
             # Do not flash because the interface already has safe guards.
@@ -241,7 +250,7 @@ def delete_user(user_id):
 
     @view.route('/edit/<int:user_id>/new-password', methods=('POST',))
     @admin_required
-    def new_password(user_id):
+    def new_password(user_id: int) -> werkzeug.Response:
         try:
             user = Journalist.query.get(user_id)
         except NoResultFound:
@@ -257,7 +266,7 @@ def new_password(user_id):
 
     @view.route('/ossec-test')
     @admin_required
-    def ossec_test():
+    def ossec_test() -> werkzeug.Response:
         current_app.logger.error('This is a test OSSEC alert')
         flash(gettext('Test alert sent. Please check your email.'),
               'notification')

securedrop/journalist_app/api.py

@@ -153,7 +153,7 @@ def single_source(source_uuid: str) -> Tuple[flask.Response, int]:
             utils.delete_collection(source.filesystem_id)
             return jsonify({'message': 'Source and submissions deleted'}), 200
         else:
-            abort(404)
+            abort(405)
 
     @api.route('/sources/<source_uuid>/add_star', methods=['POST'])
     @token_required
@@ -218,7 +218,7 @@ def single_submission(source_uuid: str, submission_uuid: str) -> Tuple[flask.Res
             utils.delete_file_object(submission)
             return jsonify({'message': 'Submission deleted'}), 200
         else:
-            abort(404)
+            abort(405)
 
     @api.route('/sources/<source_uuid>/replies', methods=['GET', 'POST'])
     @token_required
@@ -286,7 +286,7 @@ def all_source_replies(source_uuid: str) -> Tuple[flask.Response, int]:
                             'uuid': reply.uuid,
                             'filename': reply.filename}), 201
         else:
-            abort(404)
+            abort(405)
 
     @api.route('/sources/<source_uuid>/replies/<reply_uuid>',
                methods=['GET', 'DELETE'])
@@ -300,7 +300,7 @@ def single_reply(source_uuid: str, reply_uuid: str) -> Tuple[flask.Response, int
             utils.delete_file_object(reply)
             return jsonify({'message': 'Reply deleted'}), 200
         else:
-            abort(404)
+            abort(405)
 
     @api.route('/submissions', methods=['GET'])
     @token_required

securedrop/journalist_app/col.py

@@ -12,6 +12,7 @@
     send_file,
     url_for,
 )
+import werkzeug
 from flask_babel import gettext
 from sqlalchemy.orm.exc import NoResultFound
 
@@ -22,33 +23,34 @@
                                   delete_collection, col_download_unread,
                                   col_download_all, col_star, col_un_star,
                                   col_delete, mark_seen)
+from sdconfig import SDConfig
 
 
-def make_blueprint(config):
+def make_blueprint(config: SDConfig) -> Blueprint:
     view = Blueprint('col', __name__)
 
     @view.route('/add_star/<filesystem_id>', methods=('POST',))
-    def add_star(filesystem_id):
+    def add_star(filesystem_id: str) -> werkzeug.Response:
         make_star_true(filesystem_id)
         db.session.commit()
         return redirect(url_for('main.index'))
 
     @view.route("/remove_star/<filesystem_id>", methods=('POST',))
-    def remove_star(filesystem_id):
+    def remove_star(filesystem_id: str) -> werkzeug.Response:
         make_star_false(filesystem_id)
         db.session.commit()
         return redirect(url_for('main.index'))
 
     @view.route('/<filesystem_id>')
-    def col(filesystem_id):
+    def col(filesystem_id: str) -> str:
         form = ReplyForm()
         source = get_source(filesystem_id)
         source.has_key = current_app.crypto_util.get_fingerprint(filesystem_id)
         return render_template("col.html", filesystem_id=filesystem_id,
                                source=source, form=form)
 
     @view.route('/delete/<filesystem_id>', methods=('POST',))
-    def delete_single(filesystem_id):
+    def delete_single(filesystem_id: str) -> werkzeug.Response:
         """deleting a single collection from its /col page"""
         source = get_source(filesystem_id)
         try:
@@ -63,7 +65,7 @@ def delete_single(filesystem_id):
         return redirect(url_for('main.index'))
 
     @view.route('/process', methods=('POST',))
-    def process():
+    def process() -> werkzeug.Response:
         actions = {'download-unread': col_download_unread,
                    'download-all': col_download_all, 'star': col_star,
                    'un-star': col_un_star, 'delete': col_delete}
@@ -82,7 +84,7 @@ def process():
         return method(cols_selected)
 
     @view.route('/<filesystem_id>/<fn>')
-    def download_single_file(filesystem_id, fn):
+    def download_single_file(filesystem_id: str, fn: str) -> werkzeug.Response:
         """
         Marks the file being download (the file being downloaded is either a submission message,
         submission file attachement, or journalist reply) as seen by the current logged-in user and

securedrop/journalist_app/decorators.py

@@ -1,15 +1,18 @@
 # -*- coding: utf-8 -*-
+from typing import Any
 
 from flask import redirect, url_for, flash, g
 from flask_babel import gettext
 from functools import wraps
 
+from typing import Callable
+
 from journalist_app.utils import logged_in
 
 
-def admin_required(func):
+def admin_required(func: Callable) -> Callable:
     @wraps(func)
-    def wrapper(*args, **kwargs):
+    def wrapper(*args: Any, **kwargs: Any) -> Any:
         if logged_in() and g.user.is_admin:
             return func(*args, **kwargs)
         flash(gettext("Only admins can access this page."),

securedrop/journalist_app/forms.py

@@ -3,14 +3,15 @@
 from flask_babel import lazy_gettext as gettext
 from flask_wtf import FlaskForm
 from flask_wtf.file import FileField, FileAllowed, FileRequired
+from wtforms import Field
 from wtforms import (TextAreaField, StringField, BooleanField, HiddenField,
                      ValidationError)
 from wtforms.validators import InputRequired, Optional
 
 from models import Journalist
 
 
-def otp_secret_validation(form, field):
+def otp_secret_validation(form: FlaskForm, field: Field) -> None:
     strip_whitespace = field.data.replace(' ', '')
     if len(strip_whitespace) != 40:
         raise ValidationError(gettext(
@@ -20,22 +21,22 @@ def otp_secret_validation(form, field):
             )))
 
 
-def minimum_length_validation(form, field):
+def minimum_length_validation(form: FlaskForm, field: Field) -> None:
     if len(field.data) < Journalist.MIN_USERNAME_LEN:
         raise ValidationError(
             gettext('Must be at least {min_chars} '
                     'characters long.'
                     .format(min_chars=Journalist.MIN_USERNAME_LEN)))
 
 
-def name_length_validation(form, field):
+def name_length_validation(form: FlaskForm, field: Field) -> None:
     if len(field.data) > Journalist.MAX_NAME_LEN:
         raise ValidationError(gettext(
             'Cannot be longer than {max_chars} characters.'
             .format(max_chars=Journalist.MAX_NAME_LEN)))
 
 
-def check_invalid_usernames(form, field):
+def check_invalid_usernames(form: FlaskForm, field: Field) -> None:
     if field.data in Journalist.INVALID_USERNAMES:
         raise ValidationError(gettext(
             "This username is invalid because it is reserved for internal use by the software."))