vee1e/mitmproxy
1
0
Fork 0
mirror of https://github.com/vee1e/mitmproxy.git synced 2026-09-01 18:27:18 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

web: add token-based authentication for the web ui API

Browse source
This commit is contained in:
Maximilian Hils 2025-01-17 19:13:02 +01:00
parent ea0dcb0bd8
commit fa89055e19
5 changed files with 163 additions and 44 deletions
Download patch file Download diff file Expand all files Collapse all files

117
mitmproxy/tools/web/app.py
View file

@ -1,6 +1,8 @@
from __future__ import annotations
import functools
import hashlib
import hmac
import json
import logging
import os.path
@ -12,6 +14,7 @@ from collections.abc import Sequence
from io import BytesIO
from itertools import islice
from typing import ClassVar
from typing import Concatenate
import tornado.escape
import tornado.web
@ -206,7 +209,54 @@ class APIError(tornado.web.HTTPError):
pass
class RequestHandler(tornado.web.RequestHandler):
class AuthRequestHandler(tornado.web.RequestHandler):
AUTH_COOKIE_NAME = "mitmproxy-auth"
AUTH_COOKIE_VALUE = b"y"
def __init_subclass__(cls, **kwargs):
"""Automatically wrap all request handlers with `_require_auth`."""
for method in cls.SUPPORTED_METHODS:
method = method.lower()
fn = getattr(cls, method)
if fn is not tornado.web.RequestHandler._unimplemented_method:
setattr(cls, method, AuthRequestHandler._require_auth(fn))
@staticmethod
def _require_auth[**P, R](
fn: Callable[Concatenate[AuthRequestHandler, P], R],
) -> Callable[Concatenate[AuthRequestHandler, P], R | None]:
@functools.wraps(fn)
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"],
):
self.set_status(403)
self.render("login.html")
return None
self.set_signed_cookie(
self.AUTH_COOKIE_NAME,
self.AUTH_COOKIE_VALUE,
expires_days=400,
httponly=True,
samesite="Strict",
)
return fn(self, *args, **kwargs)
return wrapper
def get_current_user(self) -> bool:
return (
self.get_signed_cookie(self.AUTH_COOKIE_NAME, min_version=2)
== self.AUTH_COOKIE_VALUE
)
class RequestHandler(AuthRequestHandler):
application: Application
def write(self, chunk: str | bytes | dict | list):
@ -298,7 +348,7 @@ class FilterHelp(RequestHandler):
self.write(dict(commands=flowfilter.help))
class WebSocketEventBroadcaster(tornado.websocket.WebSocketHandler):
class WebSocketEventBroadcaster(tornado.websocket.WebSocketHandler, AuthRequestHandler):
# raise an error if inherited class doesn't specify its own instance.
connections: ClassVar[set[WebSocketEventBroadcaster]]
@ -717,6 +767,34 @@ class GZipContentAndFlowFiles(tornado.web.GZipContentEncoding):
}
handlers = [
(r"/", IndexHandler),
(r"/filter-help(?:\.json)?", FilterHelp),
(r"/updates", ClientConnection),
(r"/commands(?:\.json)?", Commands),
(r"/commands/(?P<cmd>[a-z.]+)", ExecuteCommand),
(r"/events(?:\.json)?", Events),
(r"/flows(?:\.json)?", Flows),
(r"/flows/dump", DumpFlows),
(r"/flows/resume", ResumeFlows),
(r"/flows/kill", KillFlows),
(r"/flows/(?P<flow_id>[0-9a-f\-]+)", FlowHandler),
(r"/flows/(?P<flow_id>[0-9a-f\-]+)/resume", ResumeFlow),
(r"/flows/(?P<flow_id>[0-9a-f\-]+)/kill", KillFlow),
(r"/flows/(?P<flow_id>[0-9a-f\-]+)/duplicate", DuplicateFlow),
(r"/flows/(?P<flow_id>[0-9a-f\-]+)/replay", ReplayFlow),
(r"/flows/(?P<flow_id>[0-9a-f\-]+)/revert", RevertFlow),
(r"/flows/(?P<flow_id>[0-9a-f\-]+)/(?P<message>request|response|messages)/content.data", FlowContent),
(r"/flows/(?P<flow_id>[0-9a-f\-]+)/(?P<message>request|response|messages)/content/(?P<content_view>[0-9a-zA-Z\-\_%]+)(?:\.json)?", FlowContentView),
(r"/clear", ClearAll),
(r"/options(?:\.json)?", Options),
(r"/options/save", SaveOptions),
(r"/state(?:\.json)?", State),
(r"/processes", ProcessList),
(r"/executable-icon", ProcessImage),
] # fmt: skip
class Application(tornado.web.Application):
master: mitmproxy.tools.web.master.WebMaster
@ -734,43 +812,12 @@ class Application(tornado.web.Application):
debug=debug,
autoreload=False,
transforms=[GZipContentAndFlowFiles],
auth_token=secrets.token_hex(16),
)
self.add_handlers("dns-rebind-protection", [(r"/.*", DnsRebind)])
self.add_handlers(
# make mitmweb accessible by IP only to prevent DNS rebinding.
r"^(localhost|[0-9.]+|\[[0-9a-fA-F:]+\])$",
[
(r"/", IndexHandler),
(r"/filter-help(?:\.json)?", FilterHelp),
(r"/updates", ClientConnection),
(r"/commands(?:\.json)?", Commands),
(r"/commands/(?P<cmd>[a-z.]+)", ExecuteCommand),
(r"/events(?:\.json)?", Events),
(r"/flows(?:\.json)?", Flows),
(r"/flows/dump", DumpFlows),
(r"/flows/resume", ResumeFlows),
(r"/flows/kill", KillFlows),
(r"/flows/(?P<flow_id>[0-9a-f\-]+)", FlowHandler),
(r"/flows/(?P<flow_id>[0-9a-f\-]+)/resume", ResumeFlow),
(r"/flows/(?P<flow_id>[0-9a-f\-]+)/kill", KillFlow),
(r"/flows/(?P<flow_id>[0-9a-f\-]+)/duplicate", DuplicateFlow),
(r"/flows/(?P<flow_id>[0-9a-f\-]+)/replay", ReplayFlow),
(r"/flows/(?P<flow_id>[0-9a-f\-]+)/revert", RevertFlow),
(
r"/flows/(?P<flow_id>[0-9a-f\-]+)/(?P<message>request|response|messages)/content.data",
FlowContent,
),
(
r"/flows/(?P<flow_id>[0-9a-f\-]+)/(?P<message>request|response|messages)/"
r"content/(?P<content_view>[0-9a-zA-Z\-\_%]+)(?:\.json)?",
FlowContentView,
),
(r"/clear", ClearAll),
(r"/options(?:\.json)?", Options),
(r"/options/save", SaveOptions),
(r"/state(?:\.json)?", State),
(r"/processes", ProcessList),
(r"/executable-icon", ProcessImage),
],
handlers, # type: ignore # https://github.com/tornadoweb/tornado/pull/3455
)

9
mitmproxy/tools/web/master.py
View file

@ -93,6 +93,11 @@ class WebMaster(master.Master):
},
)
@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"]}"
async def running(self):
# Register tornado with the current event loop
tornado.ioloop.IOLoop.current()
@ -109,8 +114,6 @@ class WebMaster(master.Master):
message += f"\nTry specifying a different port by using `--set web_port={self.options.web_port + 2}`."
raise OSError(e.errno, message, e.filename) from e
logger.info(
f"Web server listening at http://{self.options.web_host}:{self.options.web_port}/",
)
logger.info(f"Web server listening at {self.web_url}")
return await super().running()

30
mitmproxy/tools/web/templates/login.html Normal file
View file

@ -0,0 +1,30 @@
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<title>mitmproxy</title>
<link rel="icon" href=".{{ static_url('images/favicon.ico') }}" type="image/x-icon"/>
<meta name="viewport" content="width=device-width, initial-scale=1" />
<style>
body {
font-family: sans-serif;
display: flex;
flex-direction: column;
align-items: center;
}
input {
font-family: monospace;
}
</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">
<label>
<input type="password" name="token" size="32" placeholder="" />
</label>
<input type="submit" />
</form>
</body>
</html>

12
mitmproxy/tools/web/webaddons.py
View file

@ -1,10 +1,16 @@
from __future__ import annotations
import logging
import webbrowser
from collections.abc import Sequence
from typing import TYPE_CHECKING
from mitmproxy import ctx
from mitmproxy.tools.web.web_columns import AVAILABLE_WEB_COLUMNS
if TYPE_CHECKING:
from mitmproxy.tools.web.master import WebMaster
class WebAddon:
def load(self, loader):
@ -21,11 +27,11 @@ class WebAddon:
def running(self):
if hasattr(ctx.options, "web_open_browser") and ctx.options.web_open_browser:
web_url = f"http://{ctx.options.web_host}:{ctx.options.web_port}/"
success = open_browser(web_url)
master: WebMaster = ctx.master # type: ignore
success = open_browser(master.web_url)
if not success:
logging.info(
f"No web browser found. Please open a browser and point it to {web_url}",
f"No web browser found. Please open a browser and point it to {master.web_url}",
)

39
test/mitmproxy/tools/web/test_app.py
View file

@ -9,6 +9,7 @@ import pytest
import tornado.testing
from tornado import httpclient
from tornado import websocket
from tornado.web import create_signed_value
import mitmproxy_rs
from mitmproxy import log
@ -45,6 +46,11 @@ async def test_generated_files(filename):
)
def test_all_handlers_have_auth():
for _, handler in app.handlers:
assert issubclass(handler, app.AuthRequestHandler)
@pytest.mark.usefixtures("no_tornado_logging", "tdata")
class TestApp(tornado.testing.AsyncHTTPTestCase):
def get_app(self):
@ -73,7 +79,17 @@ class TestApp(tornado.testing.AsyncHTTPTestCase):
webapp.settings["xsrf_cookies"] = False
return webapp
@property
def auth_cookie(self) -> str:
auth_cookie = create_signed_value(
secret=self._app.settings["cookie_secret"],
name=app.AuthRequestHandler.AUTH_COOKIE_NAME,
value=app.AuthRequestHandler.AUTH_COOKIE_VALUE,
).decode()
return f"{app.AuthRequestHandler.AUTH_COOKIE_NAME}={auth_cookie}"
def fetch(self, *args, **kwargs) -> httpclient.HTTPResponse:
kwargs.setdefault("headers", {}).setdefault("Cookie", self.auth_cookie)
# tornado disallows POST without content by default.
return super().fetch(*args, **kwargs, allow_nonstandard_methods=True)
@ -379,9 +395,12 @@ class TestApp(tornado.testing.AsyncHTTPTestCase):
@tornado.testing.gen_test
def test_websocket(self):
ws_url = f"ws://localhost:{self.get_http_port()}/updates"
ws_req = httpclient.HTTPRequest(
f"ws://localhost:{self.get_http_port()}/updates",
headers={"Cookie": self.auth_cookie},
)
ws_client = yield websocket.websocket_connect(ws_url)
ws_client = yield websocket.websocket_connect(ws_req)
self.master.options.anticomp = True
r1 = yield ws_client.read_message()
@ -402,7 +421,7 @@ class TestApp(tornado.testing.AsyncHTTPTestCase):
ws_client.close()
# trigger on_close by opening a second connection.
ws_client2 = yield websocket.websocket_connect(ws_url)
ws_client2 = yield websocket.websocket_connect(ws_req)
ws_client2.close()
def test_process_list(self):
@ -440,3 +459,17 @@ class TestApp(tornado.testing.AsyncHTTPTestCase):
assert resp.code == 412
assert b"xsrf" not in resp.body
assert b"xsrf" in self.fetch("/", headers={"Sec-Fetch-Mode": "navigate"}).body
def test_unauthorized_api(self):
assert self.fetch("/", headers={"Cookie": ""}).code == 403
@tornado.testing.gen_test
def test_unauthorized_websocket(self):
try:
yield websocket.websocket_connect(
f"ws://localhost:{self.get_http_port()}/updates"
)
except httpclient.HTTPClientError as e:
assert e.code == 403
else:
assert False