20
Self Samples
3110
Cumulative
15
Lines Hit
10.68%
% of Total
5
Max Self
1319
Max Total
Intensity:
Cold
Hot
Self Time
Total Time
Show All
Hot Only
Heat
Specialization
Line
Self
Total
Code
1
"""Starlette application factory used by the CLI and embedders."""
2
3
import json
4
from pathlib import Path
5
6
from starlette.applications import Starlette
7
from starlette.concurrency import run_in_threadpool
8
from starlette.datastructures import UploadFile
9
from starlette.responses import FileResponse, JSONResponse
10
from starlette.routing import Mount, Route
11
from starlette.staticfiles import StaticFiles
12
13
from .files import IMAGE_LIMIT, TEXT_LIMIT, FileError, FileStore
14
from .security import allowed_hosts_for, is_allowed_host
15
16
STATIC = Path(__file__).parent / "static"
17
18
19
class BrowserProtection:
20
"""Check hosts against DNS rebinding and require a non-simple write header."""
21
22
def __init__(self, app, allowed_hosts):
23
self.app = app
24
self.allowed_hosts = allowed_hosts
25
26
async def __call__(self, scope, receive, send):
27
if scope["type"] == "http":
28
1
2
if not is_allowed_host(scope["headers"], self.allowed_hosts):
29
await JSONResponse(
30
{
31
"error": "Host not allowed. Configure --allowed-host to allow an additional hostname or IP address."
32
},
33
status_code=400,
34
headers={"Cache-Control": "no-store"},
35
)(scope, receive, send)
36
return
37
headers = dict(scope["headers"])
38
if (
39
1
scope["method"] not in {"GET", "HEAD", "OPTIONS"}
40
and headers.get(b"x-web-edit") != b"1"
41
):
42
await JSONResponse(
43
{"error": "Missing X-Web-Edit header."}, status_code=403
44
)(scope, receive, send)
45
return
46
47
async def secure_send(message):
48
if message["type"] == "http.response.start":
49
message.setdefault("headers", []).extend(
50
[
51
(b"x-content-type-options", b"nosniff"),
52
(b"cache-control", b"no-store"),
53
(b"referrer-policy", b"no-referrer"),
54
]
55
)
56
1
90
await send(message)
57
58
5
284
await self.app(scope, receive, secure_send)
59
else:
60
await self.app(scope, receive, send)
61
62
63
async def handle_error(request, error):
64
if isinstance(error, FileError):
65
return JSONResponse(
66
{"error": str(error), **error.details}, status_code=error.status
67
)
68
if isinstance(error, FileNotFoundError):
69
return JSONResponse({"error": "File or folder not found."}, status_code=404)
70
if isinstance(error, PermissionError):
71
return JSONResponse(
72
{"error": "Permission denied for this file or folder."}, status_code=403
73
)
74
return JSONResponse(
75
{"error": "The filesystem operation could not be completed."}, status_code=400
76
)
77
78
79
def create_app(directory=".", *, host="127.0.0.1", allowed_hosts=()):
80
trusted_hosts = allowed_hosts_for(host, allowed_hosts)
81
store = FileStore(directory)
82
83
def index(request):
84
return FileResponse(
85
STATIC / "index.html",
86
headers={
87
"Content-Security-Policy": "default-src 'self'; script-src 'self'; style-src 'self' 'unsafe-inline'; img-src 'self' data:; object-src 'none'; base-uri 'self'; frame-ancestors 'none'",
88
},
89
)
90
91
def browse(request):
92
2
1319
return JSONResponse(store.browse(request.query_params.get("path", "")))
93
94
def read_text(request):
95
1215
return JSONResponse(store.text(request.query_params.get("path", "")))
96
97
async def save_text(request):
98
# Bound JSON before decoding, allowing for JSON's escaped representation.
99
chunks = bytearray()
100
1
2
async for chunk in request.stream():
101
4
4
chunks.extend(chunk)
102
if len(chunks) > TEXT_LIMIT * 6 + 8192:
103
raise FileError("Text editing is limited to 10 MiB.", 413)
104
try:
105
1
151
body = json.loads(chunks)
106
except (ValueError, UnicodeDecodeError):
107
raise FileError("Invalid JSON.")
108
if not isinstance(body, dict) or "path" not in body or "content" not in body:
109
raise FileError("Provide path and content.")
110
result = await run_in_threadpool(
111
store.save_text, body["path"], body["content"], body.get("version")
112
)
113
2
3
return JSONResponse(result)
114
115
1
1
async def upload(request):
116
9
async with request.form(max_files=1, max_fields=2) as form:
117
file = form.get("file")
118
relative = form.get("path")
119
expected = form.get("version")
120
if not isinstance(file, UploadFile) or not isinstance(relative, str):
121
raise FileError("Provide a file and destination path.")
122
if expected is not None and not isinstance(expected, str):
123
raise FileError("Invalid version.")
124
result = await run_in_threadpool(store.write, relative, file.file, expected)
125
return JSONResponse(result, status_code=201)
126
127
def download(request):
128
1
10
path = store.regular_file(request.query_params.get("path", ""))
129
return FileResponse(
130
path, filename=path.name, media_type="application/octet-stream"
131
)
132
133
def delete(request):
134
return JSONResponse(
135
18
store.delete(
136
request.query_params.get("path", ""),
137
request.query_params.get("kind"),
138
request.query_params.get("version"),
139
)
140
)
141
142
def image(request):
143
path = store.regular_file(request.query_params.get("path", ""))
144
mime = store.image_type(path)
145
if not mime or path.stat().st_size >= IMAGE_LIMIT:
146
raise FileError(
147
"Image previews are available for images smaller than 2 MB.", 415
148
)
149
return FileResponse(
150
path,
151
media_type=mime,
152
headers={"Content-Security-Policy": "sandbox; default-src 'none'"},
153
)
154
155
app = Starlette(
156
routes=[
157
Route("/", index),
158
Route("/api/browse", browse),
159
Route("/api/text", read_text, methods=["GET"]),
160
Route("/api/text", save_text, methods=["PUT"]),
161
Route("/api/upload", upload, methods=["POST"]),
162
Route("/api/download", download),
163
Route("/api/file", delete, methods=["DELETE"]),
164
Route("/api/image", image),
165
Mount("/static", StaticFiles(directory=STATIC), name="static"),
166
],
167
exception_handlers={FileError: handle_error, OSError: handle_error},
168
)
169
app.add_middleware(BrowserProtection, allowed_hosts=trusted_hosts)
170
app.state.store = store
171
return app