Merge pull request #6300 from freedomofpress/6291-tor2web-redirect

Updated tor2web check to return a 403 status on success

Author: legoktm

securedrop/source_app/__init__.py

@@ -4,8 +4,7 @@
 import os
 import time
 import werkzeug
-from flask import (Flask, render_template, escape, flash, Markup, request, g, session,
-                   url_for)
+from flask import (Flask, render_template, request, g, session, redirect, url_for)
 from flask_babel import gettext
 from flask_assets import Environment
 from flask_wtf.csrf import CSRFProtect, CSRFError
@@ -90,28 +89,8 @@ def handle_csrf_error(e: CSRFError) -> werkzeug.Response:
     for module in [main, info, api]:
         app.register_blueprint(module.make_blueprint(config))  # type: ignore
 
-    @app.before_request
-    @ignore_static
-    def check_tor2web() -> None:
-        # ignore_static here so we only flash a single message warning
-        # about Tor2Web, corresponding to the initial page load.
-        if 'X-tor2web' in request.headers:
-            flash(
-                Markup(
-                    '<strong>{}</strong>&nbsp;{}&nbsp;<a href="{}">{}</a>'.format(
-                        escape(gettext("WARNING:")),
-                        escape(
-                            gettext(
-                                'You appear to be using Tor2Web, which does not provide anonymity.'
-                            )
-                        ),
-                        url_for('info.tor2web_warning'),
-                        escape(gettext('Why is this dangerous?')),
-                    )
-                ),
-                "banner-warning"
-            )
-
+    # before_request hooks are executed in order of declaration, so set up g object
+    # before the potential tor2web 403 response.
     @app.before_request
     @ignore_static
     def setup_g() -> Optional[werkzeug.Response]:
@@ -128,6 +107,15 @@ def setup_g() -> Optional[werkzeug.Response]:
 
         return None
 
+    @app.before_request
+    @ignore_static
+    def check_tor2web() -> Optional[werkzeug.Response]:
+        # TODO: expand header checking logic to catch modern tor2web proxies
+        if 'X-tor2web' in request.headers:
+            if request.path != url_for('info.tor2web_warning'):
+                return redirect(url_for('info.tor2web_warning'))
+        return None
+
     @app.errorhandler(404)
     def page_not_found(error: werkzeug.exceptions.HTTPException) -> Tuple[str, int]:
         return render_template('notfound.html'), 404

securedrop/source_app/info.py

@@ -1,20 +1,25 @@
 # -*- coding: utf-8 -*-
 import flask
-from flask import Blueprint, render_template, send_file, redirect, url_for
+from flask import Blueprint, render_template, send_file, redirect, url_for, flash
+from flask_babel import gettext
 import werkzeug
 
 from io import BytesIO  # noqa
 
 from encryption import EncryptionManager
 from sdconfig import SDConfig
+from source_app.utils import get_sourcev3_url
 
 
 def make_blueprint(config: SDConfig) -> Blueprint:
     view = Blueprint('info', __name__)
 
     @view.route('/tor2web-warning')
-    def tor2web_warning() -> str:
-        return render_template("tor2web-warning.html")
+    def tor2web_warning() -> flask.Response:
+        flash(gettext("Your connection is not anonymous right now!"), "error")
+        return flask.Response(
+            render_template("tor2web-warning.html", source_url=get_sourcev3_url()),
+            403)
 
     @view.route('/use-tor')
     def recommend_tor_browser() -> str:

securedrop/source_templates/tor2web-warning.html

@@ -1,9 +1,18 @@
 {% extends "base.html" %}
 {% block body %}
-<h1>{{ gettext('Why is there a warning about Tor2Web?') }}</h1>
-<p>{{ gettext('Using Tor2Web to connect to SecureDrop will not protect your anonymity.') }}</p>
-<p>{{ gettext('It could be possible for anyone monitoring your Internet traffic (your government, your Internet provider), to identify you.') }}
+<h1>{{ gettext('Proxy Service Detected') }}</h1>
+<div class="center">
+  {% include 'flashed.html' %}
+</div>
+<p>{{ gettext('You appear to be using a Tor proxy service to access SecureDrop. Proxy services do not protect your anonymity.') }}
 </p>
-<p>{{ gettext('We <strong>strongly advise</strong> you to use the <a href="www.torproject.org/projects/torbrowser.html">Tor Browser</a> instead.') }}
+<p>{{ gettext('Anyone monitoring your Internet traffic &mdash; including your government, your Internet provider, or the proxy operator &mdash; may be able to identify you.') }}
 </p>
-{% endblock %}
+<p>{{ gettext('Always use <a href="www.torproject.org/projects/torbrowser.html">Tor Browser</a> to access SecureDrop. Valid SecureDrop site addresses end with <code>.onion</code>. The correct address for this site is:') }}
+<pre>{{ source_url }}</pre>
+</p>
+<p> {{ gettext ('If you are already using Tor Browser, copy the address above and generate a new identity before you access SecureDrop.') }} {{ gettext('Click the <img src={icon} alt="" width="16" height="16">&nbsp;<b>New Identity</b> button in your Tor Browser\'s toolbar. This will clear your Tor Browser activity data on this device.').format(icon=url_for('static', filename='i/torbroom-black.png'))  }}
+</p>
+<p>{{ gettext('If there is a reasonable risk of your Internet traffic being monitored, consider connecting from a different location or network.') }}
+</p>
+{% endblock %}

securedrop/tests/test_source.py

@@ -537,38 +537,24 @@ def test_submit_sanitizes_filename(source_app):
                                         mtime=0)
 
 
-@flaky(rerun_filter=utils.flaky_filter_xfail)
-@pytest.mark.parametrize("locale", get_test_locales())
-def test_tor2web_warning_headers(config, source_app, locale):
+@pytest.mark.parametrize("test_url", ['main.index', 'main.create', 'main.submit'])
+def test_redirect_when_tor2web(config, source_app, test_url):
     with source_app.test_client() as app:
-        with InstrumentedApp(app) as ins:
-            resp = app.get(url_for('main.index', l=locale), headers=[('X-tor2web', 'encrypted')])
-            assert resp.status_code == 200
-
-            assert page_language(resp.data) == language_tag(locale)
-            msgids = [
-                "WARNING:",
-                "You appear to be using Tor2Web, which does not provide anonymity.",
-                "Why is this dangerous?",
-            ]
-            with xfail_untranslated_messages(config, locale, msgids):
-                ins.assert_message_flashed(
-                    '<strong>{}</strong>&nbsp;{}&nbsp;<a href="{}">{}</a>'.format(
-                        escape(gettext(msgids[0])),
-                        escape(gettext(msgids[1])),
-                        url_for('info.tor2web_warning'),
-                        escape(gettext(msgids[2])),
-                    ),
-                    'banner-warning'
-                )
-
+        resp = app.get(
+            url_for(test_url),
+            headers=[('X-tor2web', 'encrypted')],
+            follow_redirects=True)
+        text = resp.data.decode('utf-8')
+        assert resp.status_code == 403
+        assert "Proxy Service Detected" in text
 
 def test_tor2web_warning(source_app):
     with source_app.test_client() as app:
         resp = app.get(url_for('info.tor2web_warning'))
-        assert resp.status_code == 200
+        assert resp.status_code == 403
         text = resp.data.decode('utf-8')
-        assert "Why is there a warning about Tor2Web?" in text
+        assert "Proxy Service Detected" in text
+
 
 
 def test_why_use_tor_browser(source_app):