forked from
lass.pds.lassul.us/superconfig
This repository has no description
9.6 kB
291 lines
1"""CGI behind fcgiwrap for p.r/p.krebsco.de (paste, form, imgur) and c.r
2(cyberlocker). nginx picks the app with PASTE_APP and the item directory with
3PASTE_ITEMS. The on-disk layout is the one of the former htgen-paste,
4htgen-imgur and htgen-cyberlocker handlers, so existing items keep working.
5"""
6
7import email.utils
8import hashlib
9import json
10import os
11import re
12import shutil
13import subprocess
14import sys
15import tempfile
16import time
17import urllib.parse
18import uuid
19
20FILE = "@file@"
21EXIV2 = "@exiv2@"
22
23ENV = os.environ
24ITEMS = ENV["PASTE_ITEMS"]
25METHOD = ENV.get("REQUEST_METHOD", "GET")
26RAW_PATH, _, QUERY = ENV.get("REQUEST_URI", "/").partition("?")
27HOST = ENV.get("HTTP_HOST", "")
28STDIN = sys.stdin.buffer
29OUT = sys.stdout.buffer
30
31
32def respond(status, headers=(), body=b""):
33 """Send a CGI response; body is bytes or an open file to stream."""
34 lines = [f"Status: {status}", *(f"{k}: {v}" for k, v in headers)]
35 if isinstance(body, str):
36 body = body.encode()
37 if isinstance(body, bytes):
38 lines.append(f"Content-Length: {len(body)}")
39 OUT.write(("\r\n".join(lines) + "\r\n\r\n").encode())
40 if isinstance(body, bytes):
41 OUT.write(body)
42 else:
43 shutil.copyfileobj(body, OUT)
44
45
46def text(status, body):
47 respond(status, [("Content-Type", "text/plain; charset=UTF-8")], body)
48
49
50def send_file(path, content_type):
51 with open(path, "rb") as f:
52 size = os.fstat(f.fileno()).st_size
53 headers = [("Content-Type", content_type), ("Content-Length", size)]
54 respond("200 OK", headers, f)
55
56
57def file_type(path):
58 cmd = [FILE, "-ib", path]
59 return subprocess.run(cmd, capture_output=True, text=True).stdout.strip()
60
61
62def body_length():
63 return int(ENV.get("CONTENT_LENGTH") or 0)
64
65
66def read_body(size, head=b""):
67 """Yield head and then the next size - len(head) bytes of the body."""
68 if head:
69 yield head
70 left = size - len(head)
71 while left > 0:
72 chunk = STDIN.read(min(left, 1 << 20))
73 if not chunk:
74 break
75 left -= len(chunk)
76 yield chunk
77
78
79def save(chunks, name=None):
80 """Write chunks to ITEMS/name, replacing it, or without a name to the
81 nix-base32 sha256 of the content, kept if it exists already.
82 Returns (name, path, created)."""
83 os.makedirs(ITEMS, exist_ok=True)
84 digest = hashlib.sha256()
85 tmp = tempfile.NamedTemporaryFile(dir=ITEMS, prefix=".", delete=False)
86 try:
87 with tmp:
88 for chunk in chunks:
89 digest.update(chunk)
90 tmp.write(chunk)
91 replace = name is not None
92 name = name or nix32(digest.digest())
93 path = os.path.join(ITEMS, name)
94 created = not os.path.exists(path)
95 if replace or created:
96 os.chmod(tmp.name, 0o644)
97 os.replace(tmp.name, path)
98 return name, path, created
99 finally:
100 if os.path.exists(tmp.name):
101 os.unlink(tmp.name)
102
103
104NIX32 = "0123456789abcdfghijklmnpqrsvwxyz"
105
106
107def nix32(digest):
108 """Nix's base32 (nix-hash --to-base32)."""
109 out = ""
110 for n in reversed(range((len(digest) * 8 - 1) // 5 + 1)):
111 byte, bit = divmod(n * 5, 8)
112 c = digest[byte] >> bit
113 if byte + 1 < len(digest):
114 c |= digest[byte + 1] << (8 - bit)
115 out += NIX32[c & 0x1F]
116 return out
117
118
119def find(ident):
120 """The item whose name starts with ident (7+ chars), if exactly one."""
121 if len(ident) < 7 or not re.fullmatch("[0-9a-z]+", ident):
122 return None
123 try:
124 names = os.listdir(ITEMS)
125 except FileNotFoundError:
126 return None
127 hits = [n for n in names if n.startswith(ident) and n.isalnum()]
128 return os.path.join(ITEMS, hits[0]) if len(hits) == 1 else None
129
130
131def short(name):
132 return name[:7] if find(name[:7]) else None
133
134
135def paste():
136 if METHOD == "POST" and RAW_PATH == "/":
137 name, path, _ = save(read_body(body_length()))
138 override = ENV.get("HTTP_CONTENT_TYPE_OVERRIDE")
139 if override:
140 with open(path + ".content_type", "w") as f:
141 f.write(override)
142 refs = [f"http://{HOST}/{n}" for n in (name, short(name)) if n]
143 return text("200 OK", "".join(ref + "\n" for ref in refs))
144 item = find(RAW_PATH[1:])
145 if item and METHOD == "GET":
146 try:
147 with open(item + ".content_type") as f:
148 content_type = f.read()
149 except FileNotFoundError:
150 content_type = file_type(item)
151 return send_file(item, content_type)
152 if item and METHOD == "DELETE":
153 os.unlink(item)
154 return respond("200 OK")
155 respond("404 Not Found")
156
157
158def form():
159 """multipart/form-data upload from the p.krebsco.de page: the file is the
160 first part, after its headers and before "\r\n--boundary--\r\n"."""
161 if METHOD != "POST":
162 return respond("404 Not Found")
163 boundary = ENV.get("CONTENT_TYPE", "").partition("boundary=")[2]
164 boundary = boundary.partition(";")[0].strip('"')
165 length = body_length()
166 head = STDIN.read(min(length, 1 << 16))
167 start = head.find(b"\r\n\r\n") + 4
168 size = length - start - (len(boundary) + 8)
169 if not boundary or start < 4 or size < 0:
170 return text("400 Bad Request", "bad multipart upload\n")
171 name, _, _ = save(read_body(size, head[start:start + size]))
172 text("200 OK", f"https://{HOST}/{short(name) or name}\n")
173
174
175def imgur_response(status, data=None):
176 """https://api.imgur.com/models/basic"""
177 code = int(status.split()[0])
178 body = {"data": data, "status": code, "success": 200 <= code <= 299}
179 body = json.dumps(body, separators=(",", ":")) + "\n"
180 respond(status, [("Content-Type", "application/json; charset=UTF-8")],
181 body)
182
183
184def image_info(path):
185 cmd = [EXIV2, "print", path]
186 out = subprocess.run(cmd, capture_output=True, text=True).stdout
187 info = {}
188 for line in out.splitlines():
189 key, sep, value = line.partition(":")
190 if sep:
191 info[key.strip()] = value.strip()
192 size = re.fullmatch(r"([0-9]+)\s*x\s*([0-9]+)", info.get("Image size", ""))
193 if "MIME type" not in info or not size:
194 return None
195 return info["MIME type"], int(size[1]), int(size[2])
196
197
198def imgur():
199 query = urllib.parse.parse_qs(QUERY)
200 ident = RAW_PATH.rpartition("/")[2]
201 if METHOD == "POST" and RAW_PATH == "/image":
202 name, path, created = save(read_body(body_length()))
203 info = image_info(path)
204 if not info:
205 if created:
206 os.unlink(path)
207 return imgur_response("400 Bad Request")
208 content_type, width, height = info
209 scheme = ENV.get("REQUEST_SCHEME", "http")
210 deletehash = uuid.uuid4().hex
211 data = {
212 "id": name,
213 "title": query.get("title", [None])[0],
214 "description": query.get("description", [None])[0],
215 "datetime": time.time(),
216 "type": content_type,
217 "animated": False,
218 "width": width,
219 "height": height,
220 "size": os.path.getsize(path),
221 "views": 0,
222 "bandwidth": 0,
223 "vote": None,
224 "favorite": False,
225 "nsfw": None,
226 "section": None,
227 "account_url": None,
228 "acount_id": 0,
229 "is_ad": False,
230 "is_most_viral": False,
231 "tags": [],
232 "ad_type": 0,
233 "ad_url": "",
234 "in_gallery": False,
235 "deletehash": urllib.parse.quote(
236 f"{name}?deletehash={deletehash}", safe=""
237 ),
238 "name": "",
239 "link": f"{scheme}://{HOST}/image/{short(name) or name}",
240 }
241 os.setxattr(path, "user.deletehash", deletehash.encode())
242 os.setxattr(path, "user.data", json.dumps(data).encode())
243 return imgur_response("200 OK", data)
244 item = find(ident)
245 if item and METHOD == "GET" and RAW_PATH.startswith("/image/"):
246 try:
247 content_type = json.loads(os.getxattr(item, "user.data"))["type"]
248 except (OSError, ValueError, KeyError):
249 content_type = "application/octet-stream"
250 return send_file(item, content_type)
251 if (
252 item
253 and METHOD == "DELETE"
254 and RAW_PATH.startswith("/image/delete/")
255 ):
256 given = query.get("deletehash", [""])[0]
257 if given.encode() != os.getxattr(item, "user.deletehash"):
258 return imgur_response("401 Unauthorized")
259 os.unlink(item)
260 return imgur_response("200 OK")
261 respond("404 Not Found")
262
263
264def cyberlocker():
265 """Files by path; the raw request path, with repeated slashes collapsed,
266 percent-encoded into one file name."""
267 name = urllib.parse.quote(re.sub("/+", "/", RAW_PATH), safe="")
268 path = os.path.join(ITEMS, name)
269 if METHOD in ("POST", "PUT"):
270 save(read_body(body_length()), name)
271 return respond("204 No Content")
272 if not os.path.exists(path):
273 return respond("404 Not Found")
274 if METHOD == "GET":
275 since = ENV.get("HTTP_IF_MODIFIED_SINCE")
276 try:
277 since = since and email.utils.parsedate_to_datetime(since)
278 except (TypeError, ValueError):
279 since = None
280 if since and os.path.getmtime(path) < since.timestamp():
281 return respond("304 Not Modified")
282 return send_file(path, file_type(path))
283 if METHOD == "DELETE":
284 os.unlink(path)
285 return respond("204 No Content")
286 respond("404 Not Found")
287
288
289{"paste": paste, "form": form, "imgur": imgur, "cyberlocker": cyberlocker}[
290 ENV["PASTE_APP"]
291]()