1046
Self Samples
27486
Cumulative
71
Lines Hit
94.43%
% of Total
122
Max Self
10615
Max Total
Intensity:
Cold
Hot
Self Time
Total Time
Show All
Hot Only
Heat
Specialization
Line
Self
Total
Code
1
"""Filesystem operations, confined to one root directory."""
2
3
import hashlib
4
import mimetypes
5
import os
6
import re
7
import shutil
8
import stat
9
import tempfile
10
import threading
11
from collections import Counter
12
from difflib import SequenceMatcher
13
from pathlib import Path, PurePosixPath
14
15
IMAGE_LIMIT = 2_000_000
16
TEXT_LIMIT = 10 * 1024 * 1024
17
IMAGE_TYPES = {
18
"image/png",
19
"image/jpeg",
20
"image/gif",
21
"image/webp",
22
"image/avif",
23
"image/bmp",
24
"image/x-icon",
25
"image/vnd.microsoft.icon",
26
"image/svg+xml",
27
}
28
29
30
class FileError(Exception):
31
def __init__(self, message, status=400, **details):
32
super().__init__(message)
33
self.status = status
34
self.details = details
35
36
37
def version(path):
38
# Content hashes ignore metadata-only changes from sync tools like Dropbox.
39
6
31
with path.open("rb") as source:
40
2
534
return hashlib.file_digest(source, "sha256").hexdigest()
41
42
43
2
2
def delete_version(info):
44
"""Identify the item shown in a listing without reading file contents."""
45
26
26
value = f"{info.st_dev}:{info.st_ino}:{info.st_mtime_ns}:{info.st_size}:{stat.S_IFMT(info.st_mode)}"
46
42
42
return hashlib.sha256(value.encode()).hexdigest()
47
48
49
def decode_text(data):
50
try:
51
1
46
text = data.decode("utf-8-sig")
52
except UnicodeDecodeError:
53
raise FileError(
54
"This file is not UTF-8 text. You can download it instead.", 415
55
)
56
1898
if re.search(r"[\x00-\x08\x0b\x0c\x0e-\x1f]", text):
57
raise FileError(
58
"This looks like a binary file. You can download it instead.", 415
59
)
60
return text
61
62
63
def preserve_endings(original, edited):
64
"""Keep the endings of unchanged lines, including files with mixed endings."""
65
981
old_lines = re.split(r"\r\n|\r|\n", original)
66
994
old_endings = re.findall(r"\r\n|\r|\n", original)
67
983
new_lines = re.split(r"\r\n|\r|\n", edited)
68
1
90
default = Counter(old_endings).most_common(1)[0][0] if old_endings else "\n"
69
endings = [default] * (len(new_lines) - 1)
70
# SequenceMatcher's autojunk heuristic handles repetitive documents.
71
62
6422
for match in SequenceMatcher(None, old_lines, new_lines).get_matching_blocks():
72
7
7
for offset in range(match.size):
73
95
95
old_index, new_index = match.a + offset, match.b + offset
74
74
74
if old_index < len(old_endings) and new_index < len(endings):
75
46
46
endings[new_index] = old_endings[old_index]
76
60
222
return "".join(
77
122
122
line + (endings[i] if i < len(endings) else "")
78
40
40
for i, line in enumerate(new_lines)
79
)
80
81
82
class FileStore:
83
def __init__(self, root):
84
self.root = Path(root).resolve(strict=True)
85
if not self.root.is_dir():
86
raise ValueError("The root must be a directory")
87
self.lock = threading.RLock()
88
89
def path(self, relative, *, file=False):
90
if not isinstance(relative, str) or "\x00" in relative or "\\" in relative:
91
raise FileError("Invalid path.")
92
try:
93
relative.encode("utf-8")
94
except UnicodeEncodeError:
95
raise FileError("Invalid path.")
96
1
1
parts = PurePosixPath(relative)
97
2
3
if parts.is_absolute() or ".." in parts.parts:
98
raise FileError("Paths must stay inside the root folder.", 403)
99
if file and (not parts.parts or relative.endswith("/")):
100
raise FileError("Choose a filename.")
101
result = self.root
102
for part in parts.parts:
103
1
1
result = result / part
104
2
21
if result.is_symlink():
105
raise FileError("Symbolic links are not followed.", 403)
106
53
if not result.resolve().is_relative_to(self.root):
107
raise FileError("Paths must stay inside the root folder.", 403)
108
return result
109
110
def regular_file(self, relative):
111
1
16
path = self.path(relative, file=True)
112
if not stat.S_ISREG(path.stat().st_mode):
113
raise FileError("This is not a regular file.")
114
return path
115
116
def image_type(self, path):
117
38
45
mime = mimetypes.guess_type(path.name)[0]
118
return mime if mime in IMAGE_TYPES else None
119
120
def browse(self, relative):
121
20
folder = self.path(relative)
122
if not folder.is_dir():
123
if not folder.exists():
124
raise FileNotFoundError()
125
raise FileError("This is not a folder.")
126
entries = []
127
3
95
for child in folder.iterdir():
128
6
42
if child.name.startswith(".web-edit-"):
129
continue
130
try:
131
26
220
info = child.lstat()
132
except FileNotFoundError:
133
continue
134
2
2
kind = "directory" if stat.S_ISDIR(info.st_mode) else "file"
135
if stat.S_ISLNK(info.st_mode):
136
kind = "symlink"
137
1
1
elif not (stat.S_ISDIR(info.st_mode) or stat.S_ISREG(info.st_mode)):
138
kind = "special"
139
entries.append(
140
1
1
{
141
1
"name": child.name,
142
87
580
"path": child.relative_to(self.root).as_posix(),
143
"kind": kind,
144
"size": info.st_size,
145
"modified": info.st_mtime,
146
17
88
"delete_version": delete_version(info),
147
1
"preview": kind == "file"
148
and info.st_size < IMAGE_LIMIT
149
9
54
and bool(self.image_type(child)),
150
}
151
)
152
47
49
entries.sort(
153
1
1
key=lambda item: (
154
item["kind"] != "directory",
155
item["name"].casefold(),
156
1
1
item["name"],
157
)
158
)
159
return {
160
"path": ""
161
if folder == self.root
162
else folder.relative_to(self.root).as_posix(),
163
"root_name": self.root.name or "/",
164
"entries": entries,
165
}
166
167
def delete(self, relative, kind, expected):
168
with self.lock:
169
6
target = self.path(relative)
170
if target == self.root:
171
raise FileError("The root folder cannot be deleted.", 403)
172
if kind not in {"file", "directory"} or not expected:
173
raise FileError(
174
"Provide the item kind and version from the folder listing."
175
)
176
info = target.lstat()
177
actual_kind = "directory" if stat.S_ISDIR(info.st_mode) else "file"
178
if not (stat.S_ISDIR(info.st_mode) or stat.S_ISREG(info.st_mode)):
179
raise FileError("Only regular files and folders can be deleted.", 403)
180
if kind != actual_kind or expected != delete_version(info):
181
raise FileError(
182
"This item changed since the folder was loaded. Refresh before deleting it.",
183
409,
184
)
185
if kind == "directory":
186
# rmtree removes child symlinks themselves, never their targets.
187
shutil.rmtree(target)
188
else:
189
12
target.unlink()
190
return {"path": relative, "deleted": True}
191
192
def text(self, relative):
193
with self.lock:
194
7
path = self.regular_file(relative)
195
1
72
before = version(path)
196
if path.stat().st_size > TEXT_LIMIT:
197
raise FileError(
198
"Text editing is limited to 10 MiB. You can download this file.",
199
413,
200
)
201
5
data = path.read_bytes()
202
90
if version(path) != before:
203
raise FileError(
204
"The file changed while opening it. Please try again.", 409
205
)
206
661
return {"path": relative, "content": decode_text(data), "version": before}
207
208
def check_destination(self, relative, expected):
209
6
36
path = self.path(relative, file=True)
210
1
if path.exists():
211
if not path.is_file():
212
raise FileError(
213
"A folder or special file already exists at that path.", 409
214
)
215
1
299
current = version(path)
216
if expected != current:
217
raise FileError(
218
"This file already exists or has changed on disk.",
219
409,
220
version=current,
221
exists=True,
222
)
223
elif expected is not None:
224
raise FileError(
225
"The file was removed or moved on disk. Save your work to a new path.",
226
409,
227
)
228
return path
229
230
def write(self, relative, source, expected=None):
231
with self.lock:
232
1
118
target = self.check_destination(relative, expected)
233
2
3
target.parent.mkdir(parents=True, exist_ok=True)
234
8
self.path(relative, file=True)
235
temporary = None
236
try:
237
2
31
with tempfile.NamedTemporaryFile(
238
prefix=".web-edit-", dir=target.parent, delete=False
239
) as output:
240
temporary = Path(output.name)
241
1
28
shutil.copyfileobj(source, output, length=1024 * 1024)
242
8
output.flush()
243
20
21
os.fsync(output.fileno())
244
112
self.check_destination(relative, expected)
245
if expected is None:
246
# link is an atomic create-if-absent; never silently replace a file.
247
try:
248
77
77
os.link(temporary, target)
249
except FileExistsError:
250
raise FileError(
251
"A file appeared at this path during the upload. Please retry.",
252
409,
253
)
254
else:
255
5
5
os.chmod(temporary, stat.S_IMODE(target.stat().st_mode))
256
38
38
os.replace(temporary, target)
257
finally:
258
if temporary is not None:
259
5
9
temporary.unlink(missing_ok=True)
260
106
return {"path": relative, "version": version(target)}
261
262
def save_text(self, relative, content, expected):
263
from io import BytesIO
264
265
if not isinstance(content, str) or (
266
expected is not None and not isinstance(expected, str)
267
):
268
raise FileError(
269
"Content must be text and version must be a string or null."
270
)
271
try:
272
2
2
encoded = content.encode("utf-8")
273
except UnicodeEncodeError:
274
raise FileError("Content must be valid UTF-8 text.")
275
if len(encoded) > TEXT_LIMIT:
276
raise FileError("Text editing is limited to 10 MiB.", 413)
277
632
decode_text(encoded)
278
with self.lock:
279
107
path = self.check_destination(relative, expected)
280
if expected is not None:
281
if path.stat().st_size > TEXT_LIMIT:
282
raise FileError("Text editing is limited to 10 MiB.", 413)
283
8
original = path.read_bytes()
284
50
10615
encoded = preserve_endings(decode_text(original), content).encode(
285
"utf-8"
286
)
287
if original.startswith(b"\xef\xbb\xbf"):
288
encoded = b"\xef\xbb\xbf" + encoded
289
1
416
return self.write(relative, BytesIO(encoded), expected)