web: add password-based authentication

mhils · mhils · commit 0bd573a5995f · 2025-02-05T23:37:33.000+01:00
diff --git a/mitmproxy/tools/web/app.py b/mitmproxy/tools/web/app.py
@@ -2,7 +2,6 @@
 
 import functools
 import hashlib
-import hmac
 import json
 import logging
 import os.path
@@ -37,6 +36,7 @@
 from mitmproxy.net.http import status_codes
 from mitmproxy.tcp import TCPFlow
 from mitmproxy.tcp import TCPMessage
+from mitmproxy.tools.web.webaddons import WebAuth
 from mitmproxy.udp import UDPFlow
 from mitmproxy.udp import UDPMessage
 from mitmproxy.utils import asyncio_utils
@@ -221,6 +221,12 @@ def __init_subclass__(cls, **kwargs):
             if fn is not tornado.web.RequestHandler._unimplemented_method:
                 setattr(cls, method, AuthRequestHandler._require_auth(fn))
 
+    def auth_fail(self, invalid_password: bool) -> None:
+        """
+        Will be called when returning a 403.
+        May write a login form as the response.
+        """
+
     @staticmethod
     def _require_auth[**P, R](
         fn: Callable[Concatenate[AuthRequestHandler, P], R],
@@ -230,13 +236,10 @@ def wrapper(
             self: AuthRequestHandler, *args: P.args, **kwargs: P.kwargs
         ) -> R | None:
             if not self.current_user:
-                self.require_setting("auth_token", "AuthRequestHandler")
-                if not hmac.compare_digest(
-                    self.get_query_argument("token", default="invalid"),
-                    self.settings["auth_token"],
-                ):
+                password = self.get_argument("token", default="")
+                if not self.settings["is_valid_password"](password):
                     self.set_status(403)
-                    self.render("login.html")
+                    self.auth_fail(bool(password))
                     return None
                 self.set_signed_cookie(
                     self.AUTH_COOKIE_NAME,
@@ -332,6 +335,12 @@ def _is_fetch_mode_navigate(self) -> bool:
         # Forbid access for non-navigate fetch modes so that they can't obtain xsrf_token.
         return self.request.headers.get("Sec-Fetch-Mode", "navigate") == "navigate"
 
+    def auth_fail(self, invalid_password: bool) -> None:
+        # For mitmweb, we only write a login form for IndexHandler,
+        # which has additional Sec-Fetch-Mode protections.
+        if self._is_fetch_mode_navigate():
+            self.render("login.html", invalid_password=invalid_password)
+
     def get(self):
         # Forbid access for non-navigate fetch modes so that they can't obtain xsrf_token.
         if self._is_fetch_mode_navigate():
@@ -342,6 +351,8 @@ def get(self):
                 f"Unexpected Sec-Fetch-Mode header: {self.request.headers.get('Sec-Fetch-Mode')}",
             )
 
+    post = get  # login form
+
 
 class FilterHelp(RequestHandler):
     def get(self):
@@ -802,6 +813,7 @@ def __init__(
         self, master: mitmproxy.tools.web.master.WebMaster, debug: bool
     ) -> None:
         self.master = master
+        auth_addon: WebAuth = master.addons.get("webauth")
         super().__init__(
             default_host="dns-rebind-protection",
             template_path=os.path.join(os.path.dirname(__file__), "templates"),
@@ -812,7 +824,7 @@ def __init__(
             debug=debug,
             autoreload=False,
             transforms=[GZipContentAndFlowFiles],
-            auth_token=secrets.token_hex(16),
+            is_valid_password=auth_addon.is_valid_password,
         )
 
         self.add_handlers("dns-rebind-protection", [(r"/.*", DnsRebind)])
diff --git a/mitmproxy/tools/web/master.py b/mitmproxy/tools/web/master.py
@@ -1,5 +1,6 @@
 import errno
 import logging
+from typing import cast
 
 import tornado.httpserver
 import tornado.ioloop
@@ -41,6 +42,7 @@ def __init__(self, opts: options.Options, with_termlog: bool = True):
         self.addons.add(*addons.default_addons())
         self.addons.add(
             webaddons.WebAddon(),
+            webaddons.WebAuth(),
             intercept.Intercept(),
             readfile.ReadFileStdin(),
             static_viewer.StaticViewer(),
@@ -95,8 +97,7 @@ def _sig_servers_changed(self) -> None:
 
     @property
     def web_url(self) -> str:
-        # noinspection HttpUrlsUsage
-        return f"http://{self.options.web_host}:{self.options.web_port}/?token={self.app.settings["auth_token"]}"
+        return cast(webaddons.WebAuth, self.addons.get("webauth")).web_url
 
     async def running(self):
         # Register tornado with the current event loop
diff --git a/mitmproxy/tools/web/templates/login.html b/mitmproxy/tools/web/templates/login.html
@@ -18,12 +18,17 @@
     </style>
 </head>
 <body>
-    <h1>403 Auth Token Required</h1>
-    <p>To access mitmproxy, please enter the authentication token printed in the console below.</p>
-    <form method="GET">
+    {% if invalid_password %}
+    <h1 style="color: darkred">403 Invalid Password</h1>
+    {% else %}
+    <h1>403 Authentication Required</h1>
+    {% end %}
+    <p>To access mitmproxy, please enter the password or authentication token printed in the console.</p>
+    <form method="POST">
         <label>
             <input type="password" name="token" size="32" placeholder="" />
         </label>
+        {% module xsrf_form_html() %}
         <input type="submit" />
     </form>
 </body>
diff --git a/mitmproxy/tools/web/webaddons.py b/mitmproxy/tools/web/webaddons.py
@@ -1,16 +1,80 @@
 from __future__ import annotations
 
+import hmac
 import logging
+import secrets
 import webbrowser
 from collections.abc import Sequence
 from typing import TYPE_CHECKING
 
+import argon2
+
 from mitmproxy import ctx
+from mitmproxy import exceptions
 from mitmproxy.tools.web.web_columns import AVAILABLE_WEB_COLUMNS
 
 if TYPE_CHECKING:
     from mitmproxy.tools.web.master import WebMaster
 
+logger = logging.getLogger(__name__)
+
+
+class WebAuth:
+    _password: str
+    _hasher: argon2.PasswordHasher
+
+    def __init__(self):
+        self._password = secrets.token_hex(16)
+        self._hasher = argon2.PasswordHasher()
+
+    def load(self, loader):
+        loader.add_option(
+            "web_password",
+            str,
+            "",
+            "Password to protect the mitmweb user interface. "
+            "Values starting with `$` are interpreted as an argon2 hash, "
+            "everything else is considered a plaintext password."
+            "If not password is provided, a random token is generated on startup.",
+        )
+
+    def configure(self, updated) -> None:
+        if "web_password" in updated:
+            if ctx.options.web_password.startswith("$"):
+                try:
+                    argon2.extract_parameters(ctx.options.web_password)
+                except argon2.exceptions.InvalidHashError:
+                    raise exceptions.OptionsError(
+                        "`web_password` starts with `$`, but it's not a valid argon2 hash."
+                    )
+            elif ctx.options.web_password:
+                logger.warning(
+                    "Using a plaintext password to protect the mitmweb user interface. "
+                    "Consider using an argon2 hash for `web_password`  instead."
+                )
+            self._password = ctx.options.web_password or secrets.token_hex(16)
+
+    @property
+    def web_url(self) -> str:
+        if ctx.options.web_password:
+            auth = ""  # We don't want to print plaintext passwords (and it doesn't work for argon2 anyhow).
+        else:
+            auth = f"?token={self._password}"
+        # noinspection HttpUrlsUsage
+        return f"http://{ctx.options.web_host}:{ctx.options.web_port}/{auth}"
+
+    def is_valid_password(self, password: str) -> bool:
+        if self._password.startswith("$"):
+            try:
+                return self._hasher.verify(self._password, password)
+            except argon2.exceptions.VerificationError:
+                return False
+        else:
+            return hmac.compare_digest(
+                self._password,
+                password,
+            )
+
 
 class WebAddon:
     def load(self, loader):
diff --git a/pyproject.toml b/pyproject.toml
@@ -31,6 +31,7 @@ classifiers = [
 # It is not considered best practice to use install_requires to pin dependencies to specific versions.
 dependencies = [
     "aioquic>=1.1.0,<=1.2.0",
+    "argon2-cffi>=23.1.0,<=23.1.0",
     "asgiref>=3.2.10,<=3.8.1",
     "Brotli>=1.0,<=1.1.0",
     "certifi>=2019.9.11",  # no upper bound here to get latest CA bundle
diff --git a/test/mitmproxy/tools/web/test_app.py b/test/mitmproxy/tools/web/test_app.py
@@ -460,6 +460,18 @@ def test_xsrf_hardening_app(self):
         assert b"xsrf" not in resp.body
         assert b"xsrf" in self.fetch("/", headers={"Sec-Fetch-Mode": "navigate"}).body
 
+    def test_xsrf_hardening_login(self):
+        """Ensure that xsrf token is not provided for JS requests."""
+        resp = self.fetch("/", headers={"Sec-Fetch-Mode": "same-origin", "Cookie": ""})
+        assert resp.code == 403
+        assert b"xsrf" not in resp.body
+        assert (
+            b"xsrf"
+            in self.fetch(
+                "/", headers={"Sec-Fetch-Mode": "navigate", "Cookie": ""}
+            ).body
+        )
+
     def test_unauthorized_api(self):
         assert self.fetch("/", headers={"Cookie": ""}).code == 403
 
diff --git a/test/mitmproxy/tools/web/test_webaddons.py b/test/mitmproxy/tools/web/test_webaddons.py
@@ -0,0 +1,50 @@
+import pytest
+
+from mitmproxy.exceptions import OptionsError
+from mitmproxy.test import taddons
+from mitmproxy.tools.web import webaddons
+
+
+class TestWebAuth:
+    def test_token_auth(self):
+        a = webaddons.WebAuth()
+        with taddons.context(webaddons.WebAddon(), a) as tctx:
+            assert not a.is_valid_password("")
+            assert not a.is_valid_password("invalid")
+            assert not a.is_valid_password("test")
+
+            tctx.options.web_password = ""
+            assert not a.is_valid_password("")
+            assert not a.is_valid_password("invalid")
+            assert not a.is_valid_password("test")
+            assert a.is_valid_password(a._password)
+            assert "token" in a.web_url
+
+    def test_plaintext_auth(self, caplog):
+        a = webaddons.WebAuth()
+        with taddons.context(webaddons.WebAddon(), a) as tctx:
+            tctx.options.web_password = "test"
+            assert "Consider using an argon2 hash" in caplog.text
+            assert not a.is_valid_password("")
+            assert not a.is_valid_password("invalid")
+            assert a.is_valid_password("test")
+            assert "token" not in a.web_url
+
+    def test_argon2_auth(self, caplog):
+        a = webaddons.WebAuth()
+        with taddons.context(webaddons.WebAddon(), a) as tctx:
+            tctx.options.web_password = (
+                "$argon2id$v=19$m=8,t=1,p=1$c2FsdHNhbHQ$ieVgG5ysTJFx4k/KvmC9aQ"
+            )
+            assert not a.is_valid_password("")
+            assert not a.is_valid_password("invalid")
+            assert a.is_valid_password("test")
+            assert "token" not in a.web_url
+
+    def test_invalid_hash(self, caplog):
+        a = webaddons.WebAuth()
+        with taddons.context(webaddons.WebAddon(), a) as tctx:
+            with pytest.raises(OptionsError):
+                tctx.options.web_password = "$argon2id$"
+            assert not a.is_valid_password("")
+            assert not a.is_valid_password("test")
diff --git a/web/src/js/ducks/_options_gen.ts b/web/src/js/ducks/_options_gen.ts
@@ -95,6 +95,7 @@ export interface OptionsState {
     web_debug: boolean;
     web_host: string;
     web_open_browser: boolean;
+    web_password: string;
     web_port: number;
     web_static_viewer: string | undefined;
     websocket: boolean;
@@ -198,6 +199,7 @@ export const defaultState: OptionsState = {
     web_debug: false,
     web_host: "127.0.0.1",
     web_open_browser: true,
+    web_password: "",
     web_port: 8081,
     web_static_viewer: "",
     websocket: true,
