Compare commits
8 Commits
Author | SHA1 | Date | |
---|---|---|---|
![]() |
b2c0eed24d | ||
![]() |
3abe4f885e | ||
![]() |
f4c8252185 | ||
![]() |
daa1f8f2d5 | ||
![]() |
0901d3188a | ||
![]() |
0985d130e2 | ||
![]() |
7e0b0deb11 | ||
![]() |
e54ac3c6fd |
@ -4,6 +4,7 @@ coverage:
|
||||
default:
|
||||
target: auto
|
||||
threshold: 0.75
|
||||
informational: true
|
||||
project:
|
||||
default:
|
||||
target: auto
|
||||
|
@ -1 +1 @@
|
||||
__version__ = "22.6.0"
|
||||
__version__ = "22.6.2"
|
||||
|
@ -1315,7 +1315,7 @@ class Sanic(BaseSanic, RunnerMixin, metaclass=TouchUpMeta):
|
||||
self.config.update_config(config)
|
||||
|
||||
@property
|
||||
def asgi(self):
|
||||
def asgi(self) -> bool:
|
||||
return self.state.asgi
|
||||
|
||||
@asgi.setter
|
||||
|
@ -4,8 +4,7 @@ from functools import partial, wraps
|
||||
from inspect import getsource, signature
|
||||
from mimetypes import guess_type
|
||||
from os import path
|
||||
from pathlib import PurePath
|
||||
from re import sub
|
||||
from pathlib import Path, PurePath
|
||||
from textwrap import dedent
|
||||
from time import gmtime, strftime
|
||||
from typing import (
|
||||
@ -27,12 +26,7 @@ from sanic.base.meta import SanicMeta
|
||||
from sanic.compat import stat_async
|
||||
from sanic.constants import DEFAULT_HTTP_CONTENT_TYPE, HTTP_METHODS
|
||||
from sanic.errorpages import RESPONSE_MAPPING
|
||||
from sanic.exceptions import (
|
||||
BadRequest,
|
||||
FileNotFound,
|
||||
HeaderNotFound,
|
||||
RangeNotSatisfiable,
|
||||
)
|
||||
from sanic.exceptions import FileNotFound, HeaderNotFound, RangeNotSatisfiable
|
||||
from sanic.handlers import ContentRangeHandler
|
||||
from sanic.log import error_logger
|
||||
from sanic.models.futures import FutureRoute, FutureStatic
|
||||
@ -806,32 +800,40 @@ class RouteMixin(metaclass=SanicMeta):
|
||||
content_type=None,
|
||||
__file_uri__=None,
|
||||
):
|
||||
# Using this to determine if the URL is trying to break out of the path
|
||||
# served. os.path.realpath seems to be very slow
|
||||
if __file_uri__ and "../" in __file_uri__:
|
||||
raise BadRequest("Invalid URL")
|
||||
# Merge served directory and requested file if provided
|
||||
# Strip all / that in the beginning of the URL to help prevent python
|
||||
# from herping a derp and treating the uri as an absolute path
|
||||
root_path = file_path = file_or_directory
|
||||
if __file_uri__:
|
||||
file_path = path.join(
|
||||
file_or_directory, sub("^[/]*", "", __file_uri__)
|
||||
)
|
||||
file_path_raw = Path(unquote(file_or_directory))
|
||||
root_path = file_path = file_path_raw.resolve()
|
||||
not_found = FileNotFound(
|
||||
"File not found",
|
||||
path=file_or_directory,
|
||||
relative_url=__file_uri__,
|
||||
)
|
||||
|
||||
# URL decode the path sent by the browser otherwise we won't be able to
|
||||
# match filenames which got encoded (filenames with spaces etc)
|
||||
file_path = path.abspath(unquote(file_path))
|
||||
if not file_path.startswith(path.abspath(unquote(root_path))):
|
||||
error_logger.exception(
|
||||
f"File not found: path={file_or_directory}, "
|
||||
f"relative_url={__file_uri__}"
|
||||
)
|
||||
raise FileNotFound(
|
||||
"File not found",
|
||||
path=file_or_directory,
|
||||
relative_url=__file_uri__,
|
||||
)
|
||||
if __file_uri__:
|
||||
# Strip all / that in the beginning of the URL to help prevent
|
||||
# python from herping a derp and treating the uri as an
|
||||
# absolute path
|
||||
unquoted_file_uri = unquote(__file_uri__).lstrip("/")
|
||||
file_path_raw = Path(file_or_directory, unquoted_file_uri)
|
||||
file_path = file_path_raw.resolve()
|
||||
if (
|
||||
file_path < root_path and not file_path_raw.is_symlink()
|
||||
) or ".." in file_path_raw.parts:
|
||||
error_logger.exception(
|
||||
f"File not found: path={file_or_directory}, "
|
||||
f"relative_url={__file_uri__}"
|
||||
)
|
||||
raise not_found
|
||||
|
||||
try:
|
||||
file_path.relative_to(root_path)
|
||||
except ValueError:
|
||||
if not file_path_raw.is_symlink():
|
||||
error_logger.exception(
|
||||
f"File not found: path={file_or_directory}, "
|
||||
f"relative_url={__file_uri__}"
|
||||
)
|
||||
raise not_found
|
||||
try:
|
||||
headers = {}
|
||||
# Check if the client has been sent this file before
|
||||
@ -899,11 +901,7 @@ class RouteMixin(metaclass=SanicMeta):
|
||||
except RangeNotSatisfiable:
|
||||
raise
|
||||
except FileNotFoundError:
|
||||
raise FileNotFound(
|
||||
"File not found",
|
||||
path=file_or_directory,
|
||||
relative_url=__file_uri__,
|
||||
)
|
||||
raise not_found
|
||||
except Exception:
|
||||
error_logger.exception(
|
||||
f"Exception in static request handler: "
|
||||
|
@ -69,6 +69,7 @@ else:
|
||||
|
||||
class RunnerMixin(metaclass=SanicMeta):
|
||||
_app_registry: Dict[str, Sanic]
|
||||
asgi: bool
|
||||
config: Config
|
||||
listeners: Dict[str, List[ListenerType[Any]]]
|
||||
state: ApplicationState
|
||||
@ -525,7 +526,7 @@ class RunnerMixin(metaclass=SanicMeta):
|
||||
)
|
||||
)
|
||||
else:
|
||||
server = ""
|
||||
server = "ASGI" if self.asgi else "unknown" # type: ignore
|
||||
|
||||
display = {
|
||||
"mode": " ".join(mode),
|
||||
@ -571,8 +572,12 @@ class RunnerMixin(metaclass=SanicMeta):
|
||||
|
||||
@property
|
||||
def serve_location(self) -> str:
|
||||
server_settings = self.state.server_info[0].settings
|
||||
return self.get_server_location(server_settings)
|
||||
try:
|
||||
server_settings = self.state.server_info[0].settings
|
||||
return self.get_server_location(server_settings)
|
||||
except IndexError:
|
||||
location = "ASGI" if self.asgi else "unknown" # type: ignore
|
||||
return f"http://<{location}>"
|
||||
|
||||
@staticmethod
|
||||
def get_server_location(
|
||||
|
@ -546,3 +546,13 @@ async def test_signals_triggered(app):
|
||||
assert response.status_code == 200
|
||||
assert response.text == "test_signals_triggered"
|
||||
assert signals_triggered == signals_expected
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_asgi_serve_location(app):
|
||||
@app.get("/")
|
||||
def _request(request: Request):
|
||||
return text(request.app.serve_location)
|
||||
|
||||
_, response = await app.asgi_client.get("/")
|
||||
assert response.text == "http://<ASGI>"
|
||||
|
@ -2,6 +2,7 @@ import json as stdjson
|
||||
|
||||
from collections import namedtuple
|
||||
from pathlib import Path
|
||||
from sys import version_info
|
||||
|
||||
import pytest
|
||||
|
||||
@ -74,7 +75,10 @@ def test_full_message(client):
|
||||
"""
|
||||
)
|
||||
response = client.recv()
|
||||
assert len(response) == 151
|
||||
|
||||
# AltSvcCheck touchup removes the Alt-Svc header from the
|
||||
# response in the Python 3.9+ in this case
|
||||
assert len(response) == (151 if version_info < (3, 9) else 140)
|
||||
assert b"200 OK" in response
|
||||
|
||||
|
||||
|
@ -1,6 +1,7 @@
|
||||
import inspect
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
|
||||
from collections import Counter
|
||||
from pathlib import Path
|
||||
@ -8,7 +9,7 @@ from time import gmtime, strftime
|
||||
|
||||
import pytest
|
||||
|
||||
from sanic import text
|
||||
from sanic import Sanic, text
|
||||
from sanic.exceptions import FileNotFound
|
||||
|
||||
|
||||
@ -21,6 +22,22 @@ def static_file_directory():
|
||||
return static_directory
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def double_dotted_directory_file(static_file_directory: str):
|
||||
"""Generate double dotted directory and its files"""
|
||||
if sys.platform == "win32":
|
||||
raise Exception("Windows doesn't support double dotted directories")
|
||||
|
||||
file_path = Path(static_file_directory) / "dotted.." / "dot.txt"
|
||||
double_dotted_dir = file_path.parent
|
||||
Path.mkdir(double_dotted_dir, exist_ok=True)
|
||||
with open(file_path, "w") as f:
|
||||
f.write("DOT\n")
|
||||
yield file_path
|
||||
Path.unlink(file_path)
|
||||
Path.rmdir(double_dotted_dir)
|
||||
|
||||
|
||||
def get_file_path(static_file_directory, file_name):
|
||||
return os.path.join(static_file_directory, file_name)
|
||||
|
||||
@ -578,3 +595,43 @@ def test_resource_type_dir(app, static_file_directory):
|
||||
def test_resource_type_unknown(app, static_file_directory, caplog):
|
||||
with pytest.raises(ValueError):
|
||||
app.static("/static", static_file_directory, resource_type="unknown")
|
||||
|
||||
|
||||
@pytest.mark.skipif(
|
||||
sys.platform == "win32",
|
||||
reason="Windows does not support double dotted directories",
|
||||
)
|
||||
def test_dotted_dir_ok(
|
||||
app: Sanic, static_file_directory: str, double_dotted_directory_file: Path
|
||||
):
|
||||
app.static("/foo", static_file_directory)
|
||||
dot_relative_path = str(
|
||||
double_dotted_directory_file.relative_to(static_file_directory)
|
||||
)
|
||||
_, response = app.test_client.get("/foo/" + dot_relative_path)
|
||||
assert response.status == 200
|
||||
assert response.body == b"DOT\n"
|
||||
|
||||
|
||||
def test_breakout(app: Sanic, static_file_directory: str):
|
||||
app.static("/foo", static_file_directory)
|
||||
|
||||
_, response = app.test_client.get("/foo/..%2Ffake/server.py")
|
||||
assert response.status == 404
|
||||
|
||||
_, response = app.test_client.get("/foo/..%2Fstatic/test.file")
|
||||
assert response.status == 404
|
||||
|
||||
|
||||
@pytest.mark.skipif(
|
||||
sys.platform != "win32", reason="Block backslash on Windows only"
|
||||
)
|
||||
def test_double_backslash_prohibited_on_win32(
|
||||
app: Sanic, static_file_directory: str
|
||||
):
|
||||
app.static("/foo", static_file_directory)
|
||||
|
||||
_, response = app.test_client.get("/foo/static/..\\static/test.file")
|
||||
assert response.status == 404
|
||||
_, response = app.test_client.get("/foo/static\\../static/test.file")
|
||||
assert response.status == 404
|
||||
|
Loading…
x
Reference in New Issue
Block a user