arno/gns3-server
1
0
Fork 0
mirror of https://github.com/GNS3/gns3-server synced 2024-11-15 12:59:06 +00:00
Code Issues Releases Wiki Activity

Allow only .pcap to be downloaded from remote stream API

Browse Source 
Fix #500
This commit is contained in:
Julien Duponchelle 2016-04-26 09:52:18 +02:00
parent 75196b8a55
commit 6d6e8196d2
No known key found for this signature in database
GPG Key ID: CE8B29639E07F5E8
2 changed files with 22 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
gns3server/handlers/api/file_handler.py
View File

@ -37,6 +37,9 @@ class FileHandler:
def read(request, response):
response.enable_chunked_encoding()
if not request.json.get("location").endswith(".pcap"):
raise aiohttp.web.HTTPForbidden(text="Only .pcap file are allowed")
try:
with open(request.json.get("location"), "rb") as f:
loop = asyncio.get_event_loop()

23
tests/handlers/api/test_file.py
View File

@ -27,15 +27,15 @@ from gns3server.version import __version__
def test_stream(server, tmpdir, loop):
with open(str(tmpdir / "test"), 'w+') as f:
with open(str(tmpdir / "test.pcap"), 'w+') as f:
f.write("hello")
def go(future):
query = json.dumps({"location": str(tmpdir / "test")})
query = json.dumps({"location": str(tmpdir / "test.pcap")})
headers = {'content-type': 'application/json'}
response = yield from aiohttp.request("GET", server.get_url("/files/stream", 1), data=query, headers=headers)
response.body = yield from response.content.read(5)
with open(str(tmpdir / "test"), 'a') as f:
with open(str(tmpdir / "test.pcap"), 'a') as f:
f.write("world")
response.body += yield from response.content.read(5)
response.close()
@ -48,7 +48,8 @@ def test_stream(server, tmpdir, loop):
assert response.body == b'helloworld'
def test_stream_file_not_found(server, tmpdir, loop):
def test_stream_file_not_pcap(server, tmpdir, loop):
def go(future):
query = json.dumps({"location": str(tmpdir / "test")})
headers = {'content-type': 'application/json'}
@ -56,6 +57,20 @@ def test_stream_file_not_found(server, tmpdir, loop):
response.close()
future.set_result(response)
future = asyncio.Future()
asyncio.async(go(future))
response = loop.run_until_complete(future)
assert response.status == 403
def test_stream_file_not_found(server, tmpdir, loop):
def go(future):
query = json.dumps({"location": str(tmpdir / "test.pcap")})
headers = {'content-type': 'application/json'}
response = yield from aiohttp.request("GET", server.get_url("/files/stream", 1), data=query, headers=headers)
response.close()
future.set_result(response)
future = asyncio.Future()
asyncio.async(go(future))
response = loop.run_until_complete(future)