We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Upgrade to flask 2.0.3
*This is a good first issue for new contributors to take on, if you have any questions, please ask on the task or in our [Gitter room](https://gitter.im/freedomofpress/securedrop)!*

## Description

We currently have some type checking disabled because of a bug in flask 2.0.2 that was fixed in 2.0.3. We don't want to upgrade to 2.1.0+ yet since that's a much bigger step.

```
    # TODO: enable type checking once upstream Flask fix is available. See:
    # https://github.com/pallets/flask/issues/4295
    @app.errorhandler(CSRFError)  # type: ignore
    def handle_csrf_error(e: CSRFError) -> "Response":
```

## How to fix

You'll want to update the dependency in `securedrop-app-code-requirements.in`, probably `Flask>=2.0.3,<2.1.0`, run `make update-python3-requirements` to build the new `securedrop-app-code-requirements.txt`.

Then disable the mypy overrides (and ensure mypy passes) and test basic submission functionality using the dev container.

Note that for dependency updates we require a "[diff review](https://github.com/freedomofpress/securedrop/wiki/Dependency-specification-and-update-policies)" to be done. This will be done by a SecureDrop maintainer after you submit a PR.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `securedrop/source_app/__init__.py`
Content:
```
1 import os
2 import time
3 from os import path
4 from pathlib import Path
5 from typing import Optional, Tuple
6 
7 import i18n
8 import template_filters
9 import version
10 import werkzeug
11 from db import db
12 from flask import Flask, g, redirect, render_template, request, session, url_for
13 from flask_babel import gettext
14 from flask_wtf.csrf import CSRFError, CSRFProtect
15 from models import InstanceConfig
16 from request_that_secures_file_uploads import RequestThatSecuresFileUploads
17 from sdconfig import SDConfig
18 from source_app import api, info, main
19 from source_app.decorators import ignore_static
20 from source_app.utils import clear_session_and_redirect_to_logged_out_page
21 
22 
23 def get_logo_url(app: Flask) -> str:
24     if not app.static_folder:
25         raise FileNotFoundError
26 
27     custom_logo_filename = "i/custom_logo.png"
28     default_logo_filename = "i/logo.png"
29     custom_logo_path = Path(app.static_folder) / custom_logo_filename
30     default_logo_path = Path(app.static_folder) / default_logo_filename
31     if custom_logo_path.is_file():
32         return url_for("static", filename=custom_logo_filename)
33     elif default_logo_path.is_file():
34         return url_for("static", filename=default_logo_filename)
35 
36     raise FileNotFoundError
37 
38 
39 def create_app(config: SDConfig) -> Flask:
40     app = Flask(
41         __name__,
42         template_folder=config.SOURCE_TEMPLATES_DIR,
43         static_folder=path.join(config.SECUREDROP_ROOT, "static"),
44     )
45     app.request_class = RequestThatSecuresFileUploads
46     app.config.from_object(config.SOURCE_APP_FLASK_CONFIG_CLS)
47 
48     i18n.configure(config, app)
49 
50     @app.before_request
51     @ignore_static
52     def setup_i18n() -> None:
53         """Store i18n-related values in Flask's special g object"""
54         i18n.set_locale(config)
55 
56     # The default CSRF token expiration is 1 hour. Since large uploads can
57     # take longer than an hour over Tor, we increase the valid window to 24h.
58     app.config["WTF_CSRF_TIME_LIMIT"] = 60 * 60 * 24
59     CSRFProtect(app)
60 
61     app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
62     app.config["SQLALCHEMY_DATABASE_URI"] = config.DATABASE_URI
63     db.init_app(app)
64 
65     # TODO: enable typechecking once upstream Flask fix is available. See:
66     # https://github.com/pallets/flask/issues/4295
67     @app.errorhandler(CSRFError)  # type: ignore
68     def handle_csrf_error(e: CSRFError) -> werkzeug.Response:
69         return clear_session_and_redirect_to_logged_out_page(flask_session=session)
70 
71     app.jinja_env.trim_blocks = True
72     app.jinja_env.lstrip_blocks = True
73     app.jinja_env.globals["version"] = version.__version__
74     # Exported to source templates for being included in instructions
75     app.jinja_env.globals["submission_key_fpr"] = config.JOURNALIST_KEY
76     app.jinja_env.filters["rel_datetime_format"] = template_filters.rel_datetime_format
77     app.jinja_env.filters["nl2br"] = template_filters.nl2br
78     app.jinja_env.filters["filesizeformat"] = template_filters.filesizeformat
79     app.jinja_env.filters["html_datetime_format"] = template_filters.html_datetime_format
80     app.jinja_env.add_extension("jinja2.ext.do")
81 
82     for module in [main, info, api]:
83         app.register_blueprint(module.make_blueprint(config))  # type: ignore
84 
85     # before_request hooks are executed in order of declaration, so set up g object
86     # before the potential tor2web 403 response.
87     @app.before_request
88     @ignore_static
89     def setup_g() -> Optional[werkzeug.Response]:
90         if InstanceConfig.get_default(refresh=True).organization_name:
91             g.organization_name = (  # pylint: disable=assigning-non-slot
92                 InstanceConfig.get_default().organization_name
93             )
94         else:
95             g.organization_name = gettext("SecureDrop")  # pylint: disable=assigning-non-slot
96 
97         try:
98             g.logo = get_logo_url(app)  # pylint: disable=assigning-non-slot
99         except FileNotFoundError:
100             app.logger.error("Site logo not found.")
101 
102         return None
103 
104     @app.before_request
105     @ignore_static
106     def check_tor2web() -> Optional[werkzeug.Response]:
107         # TODO: expand header checking logic to catch modern tor2web proxies
108         if "X-tor2web" in request.headers:
109             if request.path != url_for("info.tor2web_warning"):
110                 return redirect(url_for("info.tor2web_warning"))
111         return None
112 
113     @app.errorhandler(404)
114     def page_not_found(error: werkzeug.exceptions.HTTPException) -> Tuple[str, int]:
115         return render_template("notfound.html"), 404
116 
117     @app.errorhandler(500)
118     def internal_error(error: werkzeug.exceptions.HTTPException) -> Tuple[str, int]:
119         return render_template("error.html"), 500
120 
121     # Obscure the creation time of source private keys by touching them all
122     # on startup.
123     private_keys = Path(config.GPG_KEY_DIR) / "private-keys-v1.d"
124     now = time.time()
125     for entry in os.scandir(private_keys):
126         if not entry.is_file() or not entry.name.endswith(".key"):
127             continue
128         os.utime(entry.path, times=(now, now))
129         # So the ctime is also updated
130         os.chmod(entry.path, entry.stat().st_mode)
131 
132     return app
133
```

Path: `securedrop/journalist_app/__init__.py`
Content:
```
1 # -*- coding: utf-8 -*-
2 
3 import typing
4 from datetime import datetime
5 from os import path
6 from pathlib import Path
7 
8 import i18n
9 import template_filters
10 import version
11 from db import db
12 from flask import Flask, abort, g, json, redirect, render_template, request, url_for
13 from flask_babel import gettext
14 from flask_wtf.csrf import CSRFError, CSRFProtect
15 from journalist_app import account, admin, api, col, main
16 from journalist_app.sessions import Session, session
17 from journalist_app.utils import get_source
18 from models import InstanceConfig
19 from werkzeug.exceptions import default_exceptions
20 
21 # https://www.python.org/dev/peps/pep-0484/#runtime-or-type-checking
22 if typing.TYPE_CHECKING:
23     # flake8 can not understand type annotation yet.
24     # That is why all type annotation relative import
25     # statements has to be marked as noqa.
26     # http://flake8.pycqa.org/en/latest/user/error-codes.html?highlight=f401
27     from typing import Any, Optional, Tuple, Union  # noqa: F401
28 
29     from sdconfig import SDConfig  # noqa: F401
30     from werkzeug import Response  # noqa: F401
31     from werkzeug.exceptions import HTTPException  # noqa: F401
32 
33 _insecure_views = ["main.login", "static"]
34 _insecure_api_views = ["api.get_token", "api.get_endpoints"]
35 # Timezone-naive datetime format expected by SecureDrop Client
36 API_DATETIME_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ"
37 
38 
39 def get_logo_url(app: Flask) -> str:
40     if not app.static_folder:
41         raise FileNotFoundError
42 
43     custom_logo_filename = "i/custom_logo.png"
44     default_logo_filename = "i/logo.png"
45     custom_logo_path = Path(app.static_folder) / custom_logo_filename
46     default_logo_path = Path(app.static_folder) / default_logo_filename
47     if custom_logo_path.is_file():
48         return url_for("static", filename=custom_logo_filename)
49     elif default_logo_path.is_file():
50         return url_for("static", filename=default_logo_filename)
51 
52     raise FileNotFoundError
53 
54 
55 def create_app(config: "SDConfig") -> Flask:
56     app = Flask(
57         __name__,
58         template_folder=config.JOURNALIST_TEMPLATES_DIR,
59         static_folder=path.join(config.SECUREDROP_ROOT, "static"),
60     )
61 
62     app.config.from_object(config.JOURNALIST_APP_FLASK_CONFIG_CLS)
63 
64     Session(app)
65     csrf = CSRFProtect(app)
66 
67     app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
68     app.config["SQLALCHEMY_DATABASE_URI"] = config.DATABASE_URI
69     db.init_app(app)
70 
71     class JSONEncoder(json.JSONEncoder):
72         """Custom JSON encoder to use our preferred timestamp format"""
73 
74         def default(self, obj: "Any") -> "Any":
75             if isinstance(obj, datetime):
76                 return obj.strftime(API_DATETIME_FORMAT)
77             super(JSONEncoder, self).default(obj)
78 
79     app.json_encoder = JSONEncoder
80 
81     # TODO: enable type checking once upstream Flask fix is available. See:
82     # https://github.com/pallets/flask/issues/4295
83     @app.errorhandler(CSRFError)  # type: ignore
84     def handle_csrf_error(e: CSRFError) -> "Response":
85         app.logger.error("The CSRF token is invalid.")
86         msg = gettext("You have been logged out due to inactivity.")
87         session.destroy(("error", msg), session.get("locale"))
88         return redirect(url_for("main.login"))
89 
90     def _handle_http_exception(
91         error: "HTTPException",
92     ) -> "Tuple[Union[Response, str], Optional[int]]":
93         # Workaround for no blueprint-level 404/5 error handlers, see:
94         # https://github.com/pallets/flask/issues/503#issuecomment-71383286
95         # TODO: clean up API error handling such that all except 404/5s are
96         # registered in the blueprint and 404/5s are handled at the application
97         # level.
98         handler = list(app.error_handler_spec["api"][error.code].values())[0]
99         if request.path.startswith("/api/") and handler:
100             return handler(error)  # type: ignore
101 
102         return render_template("error.html", error=error), error.code
103 
104     for code in default_exceptions:
105         app.errorhandler(code)(_handle_http_exception)
106 
107     i18n.configure(config, app)
108 
109     app.jinja_env.trim_blocks = True
110     app.jinja_env.lstrip_blocks = True
111     app.jinja_env.globals["version"] = version.__version__
112     app.jinja_env.filters["rel_datetime_format"] = template_filters.rel_datetime_format
113     app.jinja_env.filters["filesizeformat"] = template_filters.filesizeformat
114     app.jinja_env.filters["html_datetime_format"] = template_filters.html_datetime_format
115     app.jinja_env.add_extension("jinja2.ext.do")
116 
117     @app.before_request
118     def update_instance_config() -> None:
119         InstanceConfig.get_default(refresh=True)
120 
121     @app.before_request
122     def setup_g() -> "Optional[Response]":
123         """Store commonly used values in Flask's special g object"""
124 
125         i18n.set_locale(config)
126 
127         if InstanceConfig.get_default().organization_name:
128             g.organization_name = (  # pylint: disable=assigning-non-slot
129                 InstanceConfig.get_default().organization_name
130             )
131         else:
132             g.organization_name = gettext("SecureDrop")  # pylint: disable=assigning-non-slot
133 
134         try:
135             g.logo = get_logo_url(app)  # pylint: disable=assigning-non-slot
136         except FileNotFoundError:
137             app.logger.error("Site logo not found.")
138 
139         if request.path.split("/")[1] == "api":
140             if request.endpoint not in _insecure_api_views and not session.logged_in():
141                 abort(403)
142         else:
143             if request.endpoint not in _insecure_views and not session.logged_in():
144                 return redirect(url_for("main.login"))
145 
146         if request.method == "POST":
147             filesystem_id = request.form.get("filesystem_id")
148             if filesystem_id:
149                 g.filesystem_id = filesystem_id  # pylint: disable=assigning-non-slot
150                 g.source = get_source(filesystem_id)  # pylint: disable=assigning-non-slot
151 
152         return None
153 
154     app.register_blueprint(main.make_blueprint(config))
155     app.register_blueprint(account.make_blueprint(config), url_prefix="/account")
156     app.register_blueprint(admin.make_blueprint(config), url_prefix="/admin")
157     app.register_blueprint(col.make_blueprint(config), url_prefix="/col")
158     api_blueprint = api.make_blueprint(config)
159     app.register_blueprint(api_blueprint, url_prefix="/api/v1")
160     csrf.exempt(api_blueprint)
161 
162     return app
163
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/securedrop/journalist_app/__init__.py b/securedrop/journalist_app/__init__.py
--- a/securedrop/journalist_app/__init__.py
+++ b/securedrop/journalist_app/__init__.py
@@ -78,9 +78,7 @@
 
     app.json_encoder = JSONEncoder
 
-    # TODO: enable type checking once upstream Flask fix is available. See:
-    # https://github.com/pallets/flask/issues/4295
-    @app.errorhandler(CSRFError)  # type: ignore
+    @app.errorhandler(CSRFError)
     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.")
diff --git a/securedrop/source_app/__init__.py b/securedrop/source_app/__init__.py
--- a/securedrop/source_app/__init__.py
+++ b/securedrop/source_app/__init__.py
@@ -62,9 +62,7 @@
     app.config["SQLALCHEMY_DATABASE_URI"] = config.DATABASE_URI
     db.init_app(app)
 
-    # TODO: enable typechecking once upstream Flask fix is available. See:
-    # https://github.com/pallets/flask/issues/4295
-    @app.errorhandler(CSRFError)  # type: ignore
+    @app.errorhandler(CSRFError)
     def handle_csrf_error(e: CSRFError) -> werkzeug.Response:
         return clear_session_and_redirect_to_logged_out_page(flask_session=session)

{"golden_diff": "diff --git a/securedrop/journalist_app/__init__.py b/securedrop/journalist_app/__init__.py\n--- a/securedrop/journalist_app/__init__.py\n+++ b/securedrop/journalist_app/__init__.py\n@@ -78,9 +78,7 @@\n \n     app.json_encoder = JSONEncoder\n \n-    # TODO: enable type checking once upstream Flask fix is available. See:\n-    # https://github.com/pallets/flask/issues/4295\n-    @app.errorhandler(CSRFError)  # type: ignore\n+    @app.errorhandler(CSRFError)\n     def handle_csrf_error(e: CSRFError) -> \"Response\":\n         app.logger.error(\"The CSRF token is invalid.\")\n         msg = gettext(\"You have been logged out due to inactivity.\")\ndiff --git a/securedrop/source_app/__init__.py b/securedrop/source_app/__init__.py\n--- a/securedrop/source_app/__init__.py\n+++ b/securedrop/source_app/__init__.py\n@@ -62,9 +62,7 @@\n     app.config[\"SQLALCHEMY_DATABASE_URI\"] = config.DATABASE_URI\n     db.init_app(app)\n \n-    # TODO: enable typechecking once upstream Flask fix is available. See:\n-    # https://github.com/pallets/flask/issues/4295\n-    @app.errorhandler(CSRFError)  # type: ignore\n+    @app.errorhandler(CSRFError)\n     def handle_csrf_error(e: CSRFError) -> werkzeug.Response:\n         return clear_session_and_redirect_to_logged_out_page(flask_session=session)\n", "issue": "Upgrade to flask 2.0.3\n*This is a good first issue for new contributors to take on, if you have any questions, please ask on the task or in our [Gitter room](https://gitter.im/freedomofpress/securedrop)!*\r\n\r\n## Description\r\n\r\nWe currently have some type checking disabled because of a bug in flask 2.0.2 that was fixed in 2.0.3. We don't want to upgrade to 2.1.0+ yet since that's a much bigger step.\r\n\r\n```\r\n    # TODO: enable type checking once upstream Flask fix is available. See:\r\n    # https://github.com/pallets/flask/issues/4295\r\n    @app.errorhandler(CSRFError)  # type: ignore\r\n    def handle_csrf_error(e: CSRFError) -> \"Response\":\r\n```\r\n\r\n## How to fix\r\n\r\nYou'll want to update the dependency in `securedrop-app-code-requirements.in`, probably `Flask>=2.0.3,<2.1.0`, run `make update-python3-requirements` to build the new `securedrop-app-code-requirements.txt`.\r\n\r\nThen disable the mypy overrides (and ensure mypy passes) and test basic submission functionality using the dev container.\r\n\r\nNote that for dependency updates we require a \"[diff review](https://github.com/freedomofpress/securedrop/wiki/Dependency-specification-and-update-policies)\" to be done. This will be done by a SecureDrop maintainer after you submit a PR.\n", "before_files": [{"content": "import os\nimport time\nfrom os import path\nfrom pathlib import Path\nfrom typing import Optional, Tuple\n\nimport i18n\nimport template_filters\nimport version\nimport werkzeug\nfrom db import db\nfrom flask import Flask, g, redirect, render_template, request, session, url_for\nfrom flask_babel import gettext\nfrom flask_wtf.csrf import CSRFError, CSRFProtect\nfrom models import InstanceConfig\nfrom request_that_secures_file_uploads import RequestThatSecuresFileUploads\nfrom sdconfig import SDConfig\nfrom source_app import api, info, main\nfrom source_app.decorators import ignore_static\nfrom source_app.utils import clear_session_and_redirect_to_logged_out_page\n\n\ndef get_logo_url(app: Flask) -> str:\n    if not app.static_folder:\n        raise FileNotFoundError\n\n    custom_logo_filename = \"i/custom_logo.png\"\n    default_logo_filename = \"i/logo.png\"\n    custom_logo_path = Path(app.static_folder) / custom_logo_filename\n    default_logo_path = Path(app.static_folder) / default_logo_filename\n    if custom_logo_path.is_file():\n        return url_for(\"static\", filename=custom_logo_filename)\n    elif default_logo_path.is_file():\n        return url_for(\"static\", filename=default_logo_filename)\n\n    raise FileNotFoundError\n\n\ndef create_app(config: SDConfig) -> Flask:\n    app = Flask(\n        __name__,\n        template_folder=config.SOURCE_TEMPLATES_DIR,\n        static_folder=path.join(config.SECUREDROP_ROOT, \"static\"),\n    )\n    app.request_class = RequestThatSecuresFileUploads\n    app.config.from_object(config.SOURCE_APP_FLASK_CONFIG_CLS)\n\n    i18n.configure(config, app)\n\n    @app.before_request\n    @ignore_static\n    def setup_i18n() -> None:\n        \"\"\"Store i18n-related values in Flask's special g object\"\"\"\n        i18n.set_locale(config)\n\n    # The default CSRF token expiration is 1 hour. Since large uploads can\n    # take longer than an hour over Tor, we increase the valid window to 24h.\n    app.config[\"WTF_CSRF_TIME_LIMIT\"] = 60 * 60 * 24\n    CSRFProtect(app)\n\n    app.config[\"SQLALCHEMY_TRACK_MODIFICATIONS\"] = False\n    app.config[\"SQLALCHEMY_DATABASE_URI\"] = config.DATABASE_URI\n    db.init_app(app)\n\n    # TODO: enable typechecking once upstream Flask fix is available. See:\n    # https://github.com/pallets/flask/issues/4295\n    @app.errorhandler(CSRFError)  # type: ignore\n    def handle_csrf_error(e: CSRFError) -> werkzeug.Response:\n        return clear_session_and_redirect_to_logged_out_page(flask_session=session)\n\n    app.jinja_env.trim_blocks = True\n    app.jinja_env.lstrip_blocks = True\n    app.jinja_env.globals[\"version\"] = version.__version__\n    # Exported to source templates for being included in instructions\n    app.jinja_env.globals[\"submission_key_fpr\"] = config.JOURNALIST_KEY\n    app.jinja_env.filters[\"rel_datetime_format\"] = template_filters.rel_datetime_format\n    app.jinja_env.filters[\"nl2br\"] = template_filters.nl2br\n    app.jinja_env.filters[\"filesizeformat\"] = template_filters.filesizeformat\n    app.jinja_env.filters[\"html_datetime_format\"] = template_filters.html_datetime_format\n    app.jinja_env.add_extension(\"jinja2.ext.do\")\n\n    for module in [main, info, api]:\n        app.register_blueprint(module.make_blueprint(config))  # type: ignore\n\n    # before_request hooks are executed in order of declaration, so set up g object\n    # before the potential tor2web 403 response.\n    @app.before_request\n    @ignore_static\n    def setup_g() -> Optional[werkzeug.Response]:\n        if InstanceConfig.get_default(refresh=True).organization_name:\n            g.organization_name = (  # pylint: disable=assigning-non-slot\n                InstanceConfig.get_default().organization_name\n            )\n        else:\n            g.organization_name = gettext(\"SecureDrop\")  # pylint: disable=assigning-non-slot\n\n        try:\n            g.logo = get_logo_url(app)  # pylint: disable=assigning-non-slot\n        except FileNotFoundError:\n            app.logger.error(\"Site logo not found.\")\n\n        return None\n\n    @app.before_request\n    @ignore_static\n    def check_tor2web() -> Optional[werkzeug.Response]:\n        # TODO: expand header checking logic to catch modern tor2web proxies\n        if \"X-tor2web\" in request.headers:\n            if request.path != url_for(\"info.tor2web_warning\"):\n                return redirect(url_for(\"info.tor2web_warning\"))\n        return None\n\n    @app.errorhandler(404)\n    def page_not_found(error: werkzeug.exceptions.HTTPException) -> Tuple[str, int]:\n        return render_template(\"notfound.html\"), 404\n\n    @app.errorhandler(500)\n    def internal_error(error: werkzeug.exceptions.HTTPException) -> Tuple[str, int]:\n        return render_template(\"error.html\"), 500\n\n    # Obscure the creation time of source private keys by touching them all\n    # on startup.\n    private_keys = Path(config.GPG_KEY_DIR) / \"private-keys-v1.d\"\n    now = time.time()\n    for entry in os.scandir(private_keys):\n        if not entry.is_file() or not entry.name.endswith(\".key\"):\n            continue\n        os.utime(entry.path, times=(now, now))\n        # So the ctime is also updated\n        os.chmod(entry.path, entry.stat().st_mode)\n\n    return app\n", "path": "securedrop/source_app/__init__.py"}, {"content": "# -*- coding: utf-8 -*-\n\nimport typing\nfrom datetime import datetime\nfrom os import path\nfrom pathlib import Path\n\nimport i18n\nimport template_filters\nimport version\nfrom db import db\nfrom flask import Flask, abort, g, json, redirect, render_template, request, url_for\nfrom flask_babel import gettext\nfrom flask_wtf.csrf import CSRFError, CSRFProtect\nfrom journalist_app import account, admin, api, col, main\nfrom journalist_app.sessions import Session, session\nfrom journalist_app.utils import get_source\nfrom models import InstanceConfig\nfrom werkzeug.exceptions import default_exceptions\n\n# https://www.python.org/dev/peps/pep-0484/#runtime-or-type-checking\nif typing.TYPE_CHECKING:\n    # flake8 can not understand type annotation yet.\n    # That is why all type annotation relative import\n    # statements has to be marked as noqa.\n    # http://flake8.pycqa.org/en/latest/user/error-codes.html?highlight=f401\n    from typing import Any, Optional, Tuple, Union  # noqa: F401\n\n    from sdconfig import SDConfig  # noqa: F401\n    from werkzeug import Response  # noqa: F401\n    from werkzeug.exceptions import HTTPException  # noqa: F401\n\n_insecure_views = [\"main.login\", \"static\"]\n_insecure_api_views = [\"api.get_token\", \"api.get_endpoints\"]\n# Timezone-naive datetime format expected by SecureDrop Client\nAPI_DATETIME_FORMAT = \"%Y-%m-%dT%H:%M:%S.%fZ\"\n\n\ndef get_logo_url(app: Flask) -> str:\n    if not app.static_folder:\n        raise FileNotFoundError\n\n    custom_logo_filename = \"i/custom_logo.png\"\n    default_logo_filename = \"i/logo.png\"\n    custom_logo_path = Path(app.static_folder) / custom_logo_filename\n    default_logo_path = Path(app.static_folder) / default_logo_filename\n    if custom_logo_path.is_file():\n        return url_for(\"static\", filename=custom_logo_filename)\n    elif default_logo_path.is_file():\n        return url_for(\"static\", filename=default_logo_filename)\n\n    raise FileNotFoundError\n\n\ndef create_app(config: \"SDConfig\") -> Flask:\n    app = Flask(\n        __name__,\n        template_folder=config.JOURNALIST_TEMPLATES_DIR,\n        static_folder=path.join(config.SECUREDROP_ROOT, \"static\"),\n    )\n\n    app.config.from_object(config.JOURNALIST_APP_FLASK_CONFIG_CLS)\n\n    Session(app)\n    csrf = CSRFProtect(app)\n\n    app.config[\"SQLALCHEMY_TRACK_MODIFICATIONS\"] = False\n    app.config[\"SQLALCHEMY_DATABASE_URI\"] = config.DATABASE_URI\n    db.init_app(app)\n\n    class JSONEncoder(json.JSONEncoder):\n        \"\"\"Custom JSON encoder to use our preferred timestamp format\"\"\"\n\n        def default(self, obj: \"Any\") -> \"Any\":\n            if isinstance(obj, datetime):\n                return obj.strftime(API_DATETIME_FORMAT)\n            super(JSONEncoder, self).default(obj)\n\n    app.json_encoder = JSONEncoder\n\n    # TODO: enable type checking once upstream Flask fix is available. See:\n    # https://github.com/pallets/flask/issues/4295\n    @app.errorhandler(CSRFError)  # type: ignore\n    def handle_csrf_error(e: CSRFError) -> \"Response\":\n        app.logger.error(\"The CSRF token is invalid.\")\n        msg = gettext(\"You have been logged out due to inactivity.\")\n        session.destroy((\"error\", msg), session.get(\"locale\"))\n        return redirect(url_for(\"main.login\"))\n\n    def _handle_http_exception(\n        error: \"HTTPException\",\n    ) -> \"Tuple[Union[Response, str], Optional[int]]\":\n        # Workaround for no blueprint-level 404/5 error handlers, see:\n        # https://github.com/pallets/flask/issues/503#issuecomment-71383286\n        # TODO: clean up API error handling such that all except 404/5s are\n        # registered in the blueprint and 404/5s are handled at the application\n        # level.\n        handler = list(app.error_handler_spec[\"api\"][error.code].values())[0]\n        if request.path.startswith(\"/api/\") and handler:\n            return handler(error)  # type: ignore\n\n        return render_template(\"error.html\", error=error), error.code\n\n    for code in default_exceptions:\n        app.errorhandler(code)(_handle_http_exception)\n\n    i18n.configure(config, app)\n\n    app.jinja_env.trim_blocks = True\n    app.jinja_env.lstrip_blocks = True\n    app.jinja_env.globals[\"version\"] = version.__version__\n    app.jinja_env.filters[\"rel_datetime_format\"] = template_filters.rel_datetime_format\n    app.jinja_env.filters[\"filesizeformat\"] = template_filters.filesizeformat\n    app.jinja_env.filters[\"html_datetime_format\"] = template_filters.html_datetime_format\n    app.jinja_env.add_extension(\"jinja2.ext.do\")\n\n    @app.before_request\n    def update_instance_config() -> None:\n        InstanceConfig.get_default(refresh=True)\n\n    @app.before_request\n    def setup_g() -> \"Optional[Response]\":\n        \"\"\"Store commonly used values in Flask's special g object\"\"\"\n\n        i18n.set_locale(config)\n\n        if InstanceConfig.get_default().organization_name:\n            g.organization_name = (  # pylint: disable=assigning-non-slot\n                InstanceConfig.get_default().organization_name\n            )\n        else:\n            g.organization_name = gettext(\"SecureDrop\")  # pylint: disable=assigning-non-slot\n\n        try:\n            g.logo = get_logo_url(app)  # pylint: disable=assigning-non-slot\n        except FileNotFoundError:\n            app.logger.error(\"Site logo not found.\")\n\n        if request.path.split(\"/\")[1] == \"api\":\n            if request.endpoint not in _insecure_api_views and not session.logged_in():\n                abort(403)\n        else:\n            if request.endpoint not in _insecure_views and not session.logged_in():\n                return redirect(url_for(\"main.login\"))\n\n        if request.method == \"POST\":\n            filesystem_id = request.form.get(\"filesystem_id\")\n            if filesystem_id:\n                g.filesystem_id = filesystem_id  # pylint: disable=assigning-non-slot\n                g.source = get_source(filesystem_id)  # pylint: disable=assigning-non-slot\n\n        return None\n\n    app.register_blueprint(main.make_blueprint(config))\n    app.register_blueprint(account.make_blueprint(config), url_prefix=\"/account\")\n    app.register_blueprint(admin.make_blueprint(config), url_prefix=\"/admin\")\n    app.register_blueprint(col.make_blueprint(config), url_prefix=\"/col\")\n    api_blueprint = api.make_blueprint(config)\n    app.register_blueprint(api_blueprint, url_prefix=\"/api/v1\")\n    csrf.exempt(api_blueprint)\n\n    return app\n", "path": "securedrop/journalist_app/__init__.py"}], "after_files": [{"content": "import os\nimport time\nfrom os import path\nfrom pathlib import Path\nfrom typing import Optional, Tuple\n\nimport i18n\nimport template_filters\nimport version\nimport werkzeug\nfrom db import db\nfrom flask import Flask, g, redirect, render_template, request, session, url_for\nfrom flask_babel import gettext\nfrom flask_wtf.csrf import CSRFError, CSRFProtect\nfrom models import InstanceConfig\nfrom request_that_secures_file_uploads import RequestThatSecuresFileUploads\nfrom sdconfig import SDConfig\nfrom source_app import api, info, main\nfrom source_app.decorators import ignore_static\nfrom source_app.utils import clear_session_and_redirect_to_logged_out_page\n\n\ndef get_logo_url(app: Flask) -> str:\n    if not app.static_folder:\n        raise FileNotFoundError\n\n    custom_logo_filename = \"i/custom_logo.png\"\n    default_logo_filename = \"i/logo.png\"\n    custom_logo_path = Path(app.static_folder) / custom_logo_filename\n    default_logo_path = Path(app.static_folder) / default_logo_filename\n    if custom_logo_path.is_file():\n        return url_for(\"static\", filename=custom_logo_filename)\n    elif default_logo_path.is_file():\n        return url_for(\"static\", filename=default_logo_filename)\n\n    raise FileNotFoundError\n\n\ndef create_app(config: SDConfig) -> Flask:\n    app = Flask(\n        __name__,\n        template_folder=config.SOURCE_TEMPLATES_DIR,\n        static_folder=path.join(config.SECUREDROP_ROOT, \"static\"),\n    )\n    app.request_class = RequestThatSecuresFileUploads\n    app.config.from_object(config.SOURCE_APP_FLASK_CONFIG_CLS)\n\n    i18n.configure(config, app)\n\n    @app.before_request\n    @ignore_static\n    def setup_i18n() -> None:\n        \"\"\"Store i18n-related values in Flask's special g object\"\"\"\n        i18n.set_locale(config)\n\n    # The default CSRF token expiration is 1 hour. Since large uploads can\n    # take longer than an hour over Tor, we increase the valid window to 24h.\n    app.config[\"WTF_CSRF_TIME_LIMIT\"] = 60 * 60 * 24\n    CSRFProtect(app)\n\n    app.config[\"SQLALCHEMY_TRACK_MODIFICATIONS\"] = False\n    app.config[\"SQLALCHEMY_DATABASE_URI\"] = config.DATABASE_URI\n    db.init_app(app)\n\n    @app.errorhandler(CSRFError)\n    def handle_csrf_error(e: CSRFError) -> werkzeug.Response:\n        return clear_session_and_redirect_to_logged_out_page(flask_session=session)\n\n    app.jinja_env.trim_blocks = True\n    app.jinja_env.lstrip_blocks = True\n    app.jinja_env.globals[\"version\"] = version.__version__\n    # Exported to source templates for being included in instructions\n    app.jinja_env.globals[\"submission_key_fpr\"] = config.JOURNALIST_KEY\n    app.jinja_env.filters[\"rel_datetime_format\"] = template_filters.rel_datetime_format\n    app.jinja_env.filters[\"nl2br\"] = template_filters.nl2br\n    app.jinja_env.filters[\"filesizeformat\"] = template_filters.filesizeformat\n    app.jinja_env.filters[\"html_datetime_format\"] = template_filters.html_datetime_format\n    app.jinja_env.add_extension(\"jinja2.ext.do\")\n\n    for module in [main, info, api]:\n        app.register_blueprint(module.make_blueprint(config))  # type: ignore\n\n    # before_request hooks are executed in order of declaration, so set up g object\n    # before the potential tor2web 403 response.\n    @app.before_request\n    @ignore_static\n    def setup_g() -> Optional[werkzeug.Response]:\n        if InstanceConfig.get_default(refresh=True).organization_name:\n            g.organization_name = (  # pylint: disable=assigning-non-slot\n                InstanceConfig.get_default().organization_name\n            )\n        else:\n            g.organization_name = gettext(\"SecureDrop\")  # pylint: disable=assigning-non-slot\n\n        try:\n            g.logo = get_logo_url(app)  # pylint: disable=assigning-non-slot\n        except FileNotFoundError:\n            app.logger.error(\"Site logo not found.\")\n\n        return None\n\n    @app.before_request\n    @ignore_static\n    def check_tor2web() -> Optional[werkzeug.Response]:\n        # TODO: expand header checking logic to catch modern tor2web proxies\n        if \"X-tor2web\" in request.headers:\n            if request.path != url_for(\"info.tor2web_warning\"):\n                return redirect(url_for(\"info.tor2web_warning\"))\n        return None\n\n    @app.errorhandler(404)\n    def page_not_found(error: werkzeug.exceptions.HTTPException) -> Tuple[str, int]:\n        return render_template(\"notfound.html\"), 404\n\n    @app.errorhandler(500)\n    def internal_error(error: werkzeug.exceptions.HTTPException) -> Tuple[str, int]:\n        return render_template(\"error.html\"), 500\n\n    # Obscure the creation time of source private keys by touching them all\n    # on startup.\n    private_keys = Path(config.GPG_KEY_DIR) / \"private-keys-v1.d\"\n    now = time.time()\n    for entry in os.scandir(private_keys):\n        if not entry.is_file() or not entry.name.endswith(\".key\"):\n            continue\n        os.utime(entry.path, times=(now, now))\n        # So the ctime is also updated\n        os.chmod(entry.path, entry.stat().st_mode)\n\n    return app\n", "path": "securedrop/source_app/__init__.py"}, {"content": "# -*- coding: utf-8 -*-\n\nimport typing\nfrom datetime import datetime\nfrom os import path\nfrom pathlib import Path\n\nimport i18n\nimport template_filters\nimport version\nfrom db import db\nfrom flask import Flask, abort, g, json, redirect, render_template, request, url_for\nfrom flask_babel import gettext\nfrom flask_wtf.csrf import CSRFError, CSRFProtect\nfrom journalist_app import account, admin, api, col, main\nfrom journalist_app.sessions import Session, session\nfrom journalist_app.utils import get_source\nfrom models import InstanceConfig\nfrom werkzeug.exceptions import default_exceptions\n\n# https://www.python.org/dev/peps/pep-0484/#runtime-or-type-checking\nif typing.TYPE_CHECKING:\n    # flake8 can not understand type annotation yet.\n    # That is why all type annotation relative import\n    # statements has to be marked as noqa.\n    # http://flake8.pycqa.org/en/latest/user/error-codes.html?highlight=f401\n    from typing import Any, Optional, Tuple, Union  # noqa: F401\n\n    from sdconfig import SDConfig  # noqa: F401\n    from werkzeug import Response  # noqa: F401\n    from werkzeug.exceptions import HTTPException  # noqa: F401\n\n_insecure_views = [\"main.login\", \"static\"]\n_insecure_api_views = [\"api.get_token\", \"api.get_endpoints\"]\n# Timezone-naive datetime format expected by SecureDrop Client\nAPI_DATETIME_FORMAT = \"%Y-%m-%dT%H:%M:%S.%fZ\"\n\n\ndef get_logo_url(app: Flask) -> str:\n    if not app.static_folder:\n        raise FileNotFoundError\n\n    custom_logo_filename = \"i/custom_logo.png\"\n    default_logo_filename = \"i/logo.png\"\n    custom_logo_path = Path(app.static_folder) / custom_logo_filename\n    default_logo_path = Path(app.static_folder) / default_logo_filename\n    if custom_logo_path.is_file():\n        return url_for(\"static\", filename=custom_logo_filename)\n    elif default_logo_path.is_file():\n        return url_for(\"static\", filename=default_logo_filename)\n\n    raise FileNotFoundError\n\n\ndef create_app(config: \"SDConfig\") -> Flask:\n    app = Flask(\n        __name__,\n        template_folder=config.JOURNALIST_TEMPLATES_DIR,\n        static_folder=path.join(config.SECUREDROP_ROOT, \"static\"),\n    )\n\n    app.config.from_object(config.JOURNALIST_APP_FLASK_CONFIG_CLS)\n\n    Session(app)\n    csrf = CSRFProtect(app)\n\n    app.config[\"SQLALCHEMY_TRACK_MODIFICATIONS\"] = False\n    app.config[\"SQLALCHEMY_DATABASE_URI\"] = config.DATABASE_URI\n    db.init_app(app)\n\n    class JSONEncoder(json.JSONEncoder):\n        \"\"\"Custom JSON encoder to use our preferred timestamp format\"\"\"\n\n        def default(self, obj: \"Any\") -> \"Any\":\n            if isinstance(obj, datetime):\n                return obj.strftime(API_DATETIME_FORMAT)\n            super(JSONEncoder, self).default(obj)\n\n    app.json_encoder = JSONEncoder\n\n    @app.errorhandler(CSRFError)\n    def handle_csrf_error(e: CSRFError) -> \"Response\":\n        app.logger.error(\"The CSRF token is invalid.\")\n        msg = gettext(\"You have been logged out due to inactivity.\")\n        session.destroy((\"error\", msg), session.get(\"locale\"))\n        return redirect(url_for(\"main.login\"))\n\n    def _handle_http_exception(\n        error: \"HTTPException\",\n    ) -> \"Tuple[Union[Response, str], Optional[int]]\":\n        # Workaround for no blueprint-level 404/5 error handlers, see:\n        # https://github.com/pallets/flask/issues/503#issuecomment-71383286\n        # TODO: clean up API error handling such that all except 404/5s are\n        # registered in the blueprint and 404/5s are handled at the application\n        # level.\n        handler = list(app.error_handler_spec[\"api\"][error.code].values())[0]\n        if request.path.startswith(\"/api/\") and handler:\n            return handler(error)  # type: ignore\n\n        return render_template(\"error.html\", error=error), error.code\n\n    for code in default_exceptions:\n        app.errorhandler(code)(_handle_http_exception)\n\n    i18n.configure(config, app)\n\n    app.jinja_env.trim_blocks = True\n    app.jinja_env.lstrip_blocks = True\n    app.jinja_env.globals[\"version\"] = version.__version__\n    app.jinja_env.filters[\"rel_datetime_format\"] = template_filters.rel_datetime_format\n    app.jinja_env.filters[\"filesizeformat\"] = template_filters.filesizeformat\n    app.jinja_env.filters[\"html_datetime_format\"] = template_filters.html_datetime_format\n    app.jinja_env.add_extension(\"jinja2.ext.do\")\n\n    @app.before_request\n    def update_instance_config() -> None:\n        InstanceConfig.get_default(refresh=True)\n\n    @app.before_request\n    def setup_g() -> \"Optional[Response]\":\n        \"\"\"Store commonly used values in Flask's special g object\"\"\"\n\n        i18n.set_locale(config)\n\n        if InstanceConfig.get_default().organization_name:\n            g.organization_name = (  # pylint: disable=assigning-non-slot\n                InstanceConfig.get_default().organization_name\n            )\n        else:\n            g.organization_name = gettext(\"SecureDrop\")  # pylint: disable=assigning-non-slot\n\n        try:\n            g.logo = get_logo_url(app)  # pylint: disable=assigning-non-slot\n        except FileNotFoundError:\n            app.logger.error(\"Site logo not found.\")\n\n        if request.path.split(\"/\")[1] == \"api\":\n            if request.endpoint not in _insecure_api_views and not session.logged_in():\n                abort(403)\n        else:\n            if request.endpoint not in _insecure_views and not session.logged_in():\n                return redirect(url_for(\"main.login\"))\n\n        if request.method == \"POST\":\n            filesystem_id = request.form.get(\"filesystem_id\")\n            if filesystem_id:\n                g.filesystem_id = filesystem_id  # pylint: disable=assigning-non-slot\n                g.source = get_source(filesystem_id)  # pylint: disable=assigning-non-slot\n\n        return None\n\n    app.register_blueprint(main.make_blueprint(config))\n    app.register_blueprint(account.make_blueprint(config), url_prefix=\"/account\")\n    app.register_blueprint(admin.make_blueprint(config), url_prefix=\"/admin\")\n    app.register_blueprint(col.make_blueprint(config), url_prefix=\"/col\")\n    api_blueprint = api.make_blueprint(config)\n    app.register_blueprint(api_blueprint, url_prefix=\"/api/v1\")\n    csrf.exempt(api_blueprint)\n\n    return app\n", "path": "securedrop/journalist_app/__init__.py"}]}