diff --git a/mypy.ini b/mypy.ini index 624deab0641..9ed7cd6074e 100644 --- a/mypy.ini +++ b/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] diff --git a/securedrop/journalist_app/admin.py b/securedrop/journalist_app/admin.py index df4b95a8a6b..2755d165e20 100644 --- a/securedrop/journalist_app/admin.py +++ b/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,7 +62,7 @@ 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"): @@ -63,10 +70,12 @@ def update_submission_preferences(): 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/', 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/', 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//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') diff --git a/securedrop/journalist_app/api.py b/securedrop/journalist_app/api.py index e6d487e9785..78fa8945d86 100644 --- a/securedrop/journalist_app/api.py +++ b/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//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//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//replies/', 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 diff --git a/securedrop/journalist_app/col.py b/securedrop/journalist_app/col.py index a348bb3854b..383cbc2046d 100644 --- a/securedrop/journalist_app/col.py +++ b/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,25 +23,26 @@ 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/', 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/", 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('/') - 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) @@ -48,7 +50,7 @@ def col(filesystem_id): source=source, form=form) @view.route('/delete/', 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('//') - 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 diff --git a/securedrop/journalist_app/decorators.py b/securedrop/journalist_app/decorators.py index cf8a1154eff..e19e52f8886 100644 --- a/securedrop/journalist_app/decorators.py +++ b/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."), diff --git a/securedrop/journalist_app/forms.py b/securedrop/journalist_app/forms.py index dbe0fec3c12..40b30b169d4 100644 --- a/securedrop/journalist_app/forms.py +++ b/securedrop/journalist_app/forms.py @@ -3,6 +3,7 @@ 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 @@ -10,7 +11,7 @@ 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,7 +21,7 @@ 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} ' @@ -28,14 +29,14 @@ def minimum_length_validation(form, field): .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.")) diff --git a/securedrop/journalist_app/main.py b/securedrop/journalist_app/main.py index 5379f779495..13bbe7a0c4b 100644 --- a/securedrop/journalist_app/main.py +++ b/securedrop/journalist_app/main.py @@ -2,6 +2,9 @@ import os from datetime import datetime +from typing import Union + +import werkzeug from flask import (Blueprint, request, current_app, session, url_for, redirect, render_template, g, flash, abort) from flask_babel import gettext @@ -13,13 +16,14 @@ from journalist_app.forms import ReplyForm from journalist_app.utils import (validate_user, bulk_delete, download, confirm_bulk_delete, get_source) +from sdconfig import SDConfig -def make_blueprint(config): +def make_blueprint(config: SDConfig) -> Blueprint: view = Blueprint('main', __name__) @view.route('/login', methods=('GET', 'POST')) - def login(): + def login() -> Union[str, werkzeug.Response]: if request.method == 'POST': user = validate_user(request.form['username'], request.form['password'], @@ -41,14 +45,16 @@ def login(): return render_template("login.html") @view.route('/logout') - def logout(): + def logout() -> werkzeug.Response: session.pop('uid', None) session.pop('expires', None) session.pop('nonce', None) return redirect(url_for('main.index')) @view.route('/org-logo') - def select_logo(): + def select_logo() -> werkzeug.Response: + if current_app.static_folder is None: + abort(500) if os.path.exists(os.path.join(current_app.static_folder, 'i', 'custom_logo.png')): return redirect(url_for('static', filename='i/custom_logo.png')) @@ -56,7 +62,7 @@ def select_logo(): return redirect(url_for('static', filename='i/logo.png')) @view.route('/') - def index(): + def index() -> str: unstarred = [] starred = [] @@ -82,7 +88,7 @@ def index(): starred=starred) @view.route('/reply', methods=('POST',)) - def reply(): + def reply() -> werkzeug.Response: """Attempt to send a Reply from a Journalist to a Source. Empty messages are rejected, and an informative error message is flashed on the client. In the case of unexpected errors involving database @@ -138,14 +144,14 @@ def reply(): return redirect(url_for('col.col', filesystem_id=g.filesystem_id)) @view.route('/flag', methods=('POST',)) - def flag(): + def flag() -> str: g.source.flagged = True db.session.commit() return render_template('flag.html', filesystem_id=g.filesystem_id, codename=g.source.journalist_designation) @view.route('/bulk', methods=('POST',)) - def bulk(): + def bulk() -> Union[str, werkzeug.Response]: action = request.form['action'] doc_names_selected = request.form.getlist('doc_names_selected') @@ -171,7 +177,7 @@ def bulk(): abort(400) @view.route('/download_unread/') - def download_unread_filesystem_id(filesystem_id): + def download_unread_filesystem_id(filesystem_id: str) -> werkzeug.Response: id = Source.query.filter(Source.filesystem_id == filesystem_id) \ .filter_by(deleted_at=None).one().id submissions = Submission.query.filter(Submission.source_id == id).all() diff --git a/securedrop/journalist_app/utils.py b/securedrop/journalist_app/utils.py index 32af2e512b2..7c577f54b27 100644 --- a/securedrop/journalist_app/utils.py +++ b/securedrop/journalist_app/utils.py @@ -190,7 +190,7 @@ def mark_seen(targets: List[Union[Submission, Reply]], user: Journalist) -> None raise -def download(zip_basename: str, submissions: List[Union[Submission, Reply]]) -> werkzeug.Response: +def download(zip_basename: str, submissions: List[Union[Submission, Reply]]) -> flask.Response: """Send client contents of ZIP-file *zip_basename*-.zip containing *submissions*. The ZIP-file, being a :class:`tempfile.NamedTemporaryFile`, is stored on disk only