Produce a better error in case of malformed JSON

$ curl localhost:8777 -X POST -i -d '{"action": "version", "version": 6},' && echo ␄
HTTP/1.1 200 OK
Content-Type: text/json
Access-Control-Allow-Origin: http://localhost
Access-Control-Allow-Headers: *
Content-Length: 67

{"result": null, "error": "Extra data: line 1 column 36 (char 35)"}␄

$ curl localhost:8777 -X POST -i -d '{"action": "version", "version": 6},' -H "Origin: foo" && echo ␄
HTTP/1.1 403 Forbidden
Access-Control-Allow-Origin: http://localhost
Access-Control-Allow-Headers: *

␄
This commit is contained in:
oakkitten 2022-04-20 23:00:24 +01:00
parent 8a84db971a
commit bffbb051f2
3 changed files with 67 additions and 27 deletions

View File

@ -43,6 +43,7 @@ from anki.notes import Note
from anki.errors import NotFoundError from anki.errors import NotFoundError
from aqt.qt import Qt, QTimer, QMessageBox, QCheckBox from aqt.qt import Qt, QTimer, QMessageBox, QCheckBox
from .web import format_exception_reply, format_success_reply
from .edit import Edit from .edit import Edit
from . import web, util from . import web, util
@ -99,7 +100,6 @@ class AnkiConnect:
version = request.get('version', 4) version = request.get('version', 4)
params = request.get('params', {}) params = request.get('params', {})
key = request.get('key') key = request.get('key')
reply = {'result': None, 'error': None}
try: try:
if key != util.setting('apiKey') and name != 'requestPermission': if key != util.setting('apiKey') and name != 'requestPermission':
@ -126,14 +126,12 @@ class AnkiConnect:
if method is None: if method is None:
raise Exception('unsupported action') raise Exception('unsupported action')
else:
reply['result'] = methodInst(**params)
if version <= 4: api_return_value = methodInst(**params)
reply = reply['result'] reply = format_success_reply(version, api_return_value)
except Exception as e: except Exception as e:
reply['error'] = str(e) reply = format_exception_reply(version, e)
self.logEvent('reply', reply) self.logEvent('reply', reply)
return reply return reply

View File

@ -175,27 +175,29 @@ class WebServer:
return self.buildResponse(headers, body) return self.buildResponse(headers, body)
paramsError = False
try: try:
params = json.loads(req.body.decode('utf-8')) params = json.loads(req.body.decode('utf-8'))
except ValueError: except ValueError as e:
body = json.dumps(None).encode('utf-8') if allowed:
paramsError = True if len(req.body) == 0:
body = f"AnkiConnect v.{util.setting('apiVersion')}".encode()
if allowed or not paramsError and params.get('action', '') == 'requestPermission': else:
if len(req.body) == 0: reply = format_exception_reply(util.setting('apiVersion'), e)
body = 'AnkiConnect v.{}'.format(util.setting('apiVersion')).encode('utf-8') body = json.dumps(reply).encode('utf-8')
headers = self.buildHeaders(corsOrigin, body)
return self.buildResponse(headers, body)
else: else:
if params.get('action', '') == 'requestPermission': params = {} # trigger the 403 response below
params['params'] = params.get('params', {})
params['params']['allowed'] = allowed if allowed or params.get('action', '') == 'requestPermission':
params['params']['origin'] = b'origin' in req.headers and req.headers[b'origin'].decode() or '' if params.get('action', '') == 'requestPermission':
if not allowed : params['params'] = params.get('params', {})
corsOrigin = params['params']['origin'] params['params']['allowed'] = allowed
params['params']['origin'] = b'origin' in req.headers and req.headers[b'origin'].decode() or ''
if not allowed :
corsOrigin = params['params']['origin']
body = json.dumps(self.handler(params)).encode('utf-8') body = json.dumps(self.handler(params)).encode('utf-8')
headers = self.buildHeaders(corsOrigin, body) headers = self.buildHeaders(corsOrigin, body)
else : else :
headers = [ headers = [
@ -273,3 +275,14 @@ class WebServer:
client.close() client.close()
self.clients = [] self.clients = []
def format_success_reply(api_version, result):
if api_version <= 4:
return result
else:
return {"result": result, "error": None}
def format_exception_reply(_api_version, exception):
return {"result": None, "error": str(exception)}

View File

@ -46,11 +46,14 @@ class Client:
return {"action": action, "params": params, "version": 6} return {"action": action, "params": params, "version": 6}
def send_request(self, action, **params): def send_request(self, action, **params):
request_url = f"http://localhost:{self.port}"
request_data = self.make_request(action, **params) request_data = self.make_request(action, **params)
request_json = json.dumps(request_data).encode("utf-8") json_bytes = json.dumps(request_data).encode("utf-8")
request = urllib.request.Request(request_url, request_json) return json.loads(self.send_bytes(json_bytes))
response = json.load(urllib.request.urlopen(request))
def send_bytes(self, bytes, headers={}): # noqa
request_url = f"http://localhost:{self.port}"
request = urllib.request.Request(request_url, bytes, headers)
response = urllib.request.urlopen(request).read()
return response return response
def wait_for_web_server_to_come_live(self, at_most_seconds=30): def wait_for_web_server_to_come_live(self, at_most_seconds=30):
@ -137,6 +140,11 @@ def test_multi_request(external_anki):
} }
def test_request_with_empty_body_returns_version_banner(external_anki):
response = external_anki.send_bytes(b"")
assert response == b"AnkiConnect v.6"
def test_failing_request_due_to_bad_arguments(external_anki): def test_failing_request_due_to_bad_arguments(external_anki):
response = external_anki.send_request("addNote", bad="request") response = external_anki.send_request("addNote", bad="request")
assert response["result"] is None assert response["result"] is None
@ -147,3 +155,24 @@ def test_failing_request_due_to_anki_raising_exception(external_anki):
response = external_anki.send_request("suspend", cards=[-123]) response = external_anki.send_request("suspend", cards=[-123])
assert response["result"] is None assert response["result"] is None
assert "Card was not found" in response["error"] assert "Card was not found" in response["error"]
def test_failing_request_due_to_bad_encoding(external_anki):
response = json.loads(external_anki.send_bytes(b"\xe7\x8c"))
assert response["result"] is None
assert "can't decode" in response["error"]
def test_failing_request_due_to_bad_json(external_anki):
response = json.loads(external_anki.send_bytes(b'{1: 2}'))
assert response["result"] is None
assert "in double quotes" in response["error"]
def test_403_in_case_of_disallowed_origin(external_anki):
with pytest.raises(urllib.error.HTTPError, match="403"): # good request/json
json_bytes = json.dumps(Client.make_request("version")).encode("utf-8")
external_anki.send_bytes(json_bytes, headers={b"origin": b"foo"})
with pytest.raises(urllib.error.HTTPError, match="403"): # bad json
external_anki.send_bytes(b'{1: 2}', headers={b"origin": b"foo"})