mirror of
https://github.com/Grasscutters/mitmproxy.git
synced 2025-01-30 14:58:38 +00:00
Merge pull request #2395 from MatthewShao/mitmweb-options
[WIP] Add RESTful API for mitmweb option
This commit is contained in:
commit
bde64746a9
@ -401,21 +401,36 @@ def dump_defaults(opts):
|
||||
if o.choices:
|
||||
txt += " Valid values are %s." % ", ".join(repr(c) for c in o.choices)
|
||||
else:
|
||||
if o.typespec in (str, int, bool):
|
||||
t = o.typespec.__name__
|
||||
elif o.typespec == typing.Optional[str]:
|
||||
t = "optional str"
|
||||
elif o.typespec == typing.Sequence[str]:
|
||||
t = "sequence of str"
|
||||
else: # pragma: no cover
|
||||
raise NotImplementedError
|
||||
t = typecheck.typespec_to_str(o.typespec)
|
||||
txt += " Type %s." % t
|
||||
|
||||
txt = "\n".join(textwrap.wrap(txt))
|
||||
s.yaml_set_comment_before_after_key(k, before = "\n" + txt)
|
||||
s.yaml_set_comment_before_after_key(k, before="\n" + txt)
|
||||
return ruamel.yaml.round_trip_dump(s)
|
||||
|
||||
|
||||
def dump_dicts(opts):
|
||||
"""
|
||||
Dumps the options into a list of dict object.
|
||||
|
||||
Return: A list like: [ { name: "anticache", type: "bool", default: false, value: true, help: "help text"} ]
|
||||
"""
|
||||
options_list = []
|
||||
for k in sorted(opts.keys()):
|
||||
o = opts._options[k]
|
||||
t = typecheck.typespec_to_str(o.typespec)
|
||||
option = {
|
||||
'name': k,
|
||||
'type': t,
|
||||
'default': o.default,
|
||||
'value': o.current(),
|
||||
'help': o.help,
|
||||
'choices': o.choices
|
||||
}
|
||||
options_list.append(option)
|
||||
return options_list
|
||||
|
||||
|
||||
def parse(text):
|
||||
if not text:
|
||||
return {}
|
||||
|
@ -17,6 +17,7 @@ from mitmproxy import http
|
||||
from mitmproxy import io
|
||||
from mitmproxy import log
|
||||
from mitmproxy import version
|
||||
from mitmproxy import optmanager
|
||||
import mitmproxy.tools.web.master # noqa
|
||||
|
||||
|
||||
@ -438,6 +439,18 @@ class Settings(RequestHandler):
|
||||
self.master.options.update(**update)
|
||||
|
||||
|
||||
class Options(RequestHandler):
|
||||
def get(self):
|
||||
self.write(optmanager.dump_dicts(self.master.options))
|
||||
|
||||
def put(self):
|
||||
update = self.json
|
||||
try:
|
||||
self.master.options.update(**update)
|
||||
except (KeyError, TypeError) as err:
|
||||
raise APIError(400, "{}".format(err))
|
||||
|
||||
|
||||
class Application(tornado.web.Application):
|
||||
def __init__(self, master, debug):
|
||||
self.master = master
|
||||
@ -462,6 +475,7 @@ class Application(tornado.web.Application):
|
||||
FlowContentView),
|
||||
(r"/settings", Settings),
|
||||
(r"/clear", ClearAll),
|
||||
(r"/options", Options)
|
||||
]
|
||||
settings = dict(
|
||||
template_path=os.path.join(os.path.dirname(__file__), "templates"),
|
||||
|
@ -98,3 +98,15 @@ def check_option_type(name: str, value: typing.Any, typeinfo: typing.Any) -> Non
|
||||
return
|
||||
elif not isinstance(value, typeinfo):
|
||||
raise e
|
||||
|
||||
|
||||
def typespec_to_str(typespec: typing.Any) -> str:
|
||||
if typespec in (str, int, bool):
|
||||
t = typespec.__name__
|
||||
elif typespec == typing.Optional[str]:
|
||||
t = 'optional str'
|
||||
elif typespec == typing.Sequence[str]:
|
||||
t = 'sequence of str'
|
||||
else:
|
||||
raise NotImplementedError
|
||||
return t
|
||||
|
@ -338,6 +338,11 @@ def test_dump_defaults():
|
||||
assert optmanager.dump_defaults(o)
|
||||
|
||||
|
||||
def test_dump_dicts():
|
||||
o = options.Options()
|
||||
assert optmanager.dump_dicts(o)
|
||||
|
||||
|
||||
class TTypes(optmanager.OptManager):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
|
@ -253,6 +253,16 @@ class TestApp(tornado.testing.AsyncHTTPTestCase):
|
||||
assert self.put_json("/settings", {"anticache": True}).code == 200
|
||||
assert self.put_json("/settings", {"wtf": True}).code == 400
|
||||
|
||||
def test_options(self):
|
||||
j = json(self.fetch("/options"))
|
||||
assert type(j) == list
|
||||
assert type(j[0]) == dict
|
||||
|
||||
def test_option_update(self):
|
||||
assert self.put_json("/options", {"anticache": True}).code == 200
|
||||
assert self.put_json("/options", {"wtf": True}).code == 400
|
||||
assert self.put_json("/options", {"anticache": "foo"}).code == 400
|
||||
|
||||
def test_err(self):
|
||||
with mock.patch("mitmproxy.tools.web.app.IndexHandler.get") as f:
|
||||
f.side_effect = RuntimeError
|
||||
|
@ -111,3 +111,11 @@ def test_check_command_type():
|
||||
m.__str__ = lambda self: "typing.Union"
|
||||
m.__union_params__ = (int,)
|
||||
assert not typecheck.check_command_type([22], m)
|
||||
|
||||
|
||||
def test_typesec_to_str():
|
||||
assert(typecheck.typespec_to_str(str)) == "str"
|
||||
assert(typecheck.typespec_to_str(typing.Sequence[str])) == "sequence of str"
|
||||
assert(typecheck.typespec_to_str(typing.Optional[str])) == "optional str"
|
||||
with pytest.raises(NotImplementedError):
|
||||
typecheck.typespec_to_str(dict)
|
||||
|
Loading…
Reference in New Issue
Block a user