This repository has no description
9.5 kB
262 lines
1"""HTTP upload endpoint that writes files under /var/lib/ipfs/download and pins them.
2
3Supports resumable uploads via `Content-Range: bytes <start>-<end>/<total>`.
4The staging file is `<STAGING_DIR>/<sha256(rel)>.upload` so the staging path
5is deterministic from the destination path; clients resume by re-POSTing
6the same URL.
7
8Protocol:
9 POST /<path> body=N bytes, no Content-Range
10 One-shot upload: equivalent to a single chunk covering the whole file.
11
12 POST /<path> body=N bytes, Content-Range: bytes <start>-<end>/<total>
13 Chunked upload. `start` must equal the current staging size, or 0
14 to truncate and start fresh. When the staging size reaches `total`,
15 the staging file is moved to /var/lib/ipfs/download/<path>; the
16 ipfs-pin-watcher then pins it (it is the sole pinner). This handler
17 only computes the CID (`ipfs add --only-hash`) to return. Returns 200
18 with the CID on completion, 308 with `Upload-Offset: <size>` while
19 incomplete.
20
21 POST /<path> body=0, no Content-Range
22 Status check. Returns 200 with `Upload-Offset: <size>` if a
23 staging file exists, 404 otherwise.
24
25URL path segments are restricted to [A-Za-z0-9._-]. Re-uploading the same
26path overwrites the file; the previous CID is left in IPFS but unreferenced
27by this service.
28"""
29
30from __future__ import annotations
31
32import hashlib
33import http.server
34import os
35import re
36import socketserver
37import subprocess
38import sys
39import threading
40
41DOWNLOAD_ROOT = os.environ.get("DOWNLOAD_ROOT", "/var/lib/ipfs/download")
42STAGING_DIR = os.environ.get("STAGING_DIR", "/var/lib/ipfs/download/incoming/uploader")
43IPFS_BIN = os.environ.get("IPFS_BIN", "ipfs")
44CHUNK = 1 << 20 # 1 MiB
45_CONTENT_RANGE = re.compile(r"^bytes (\d+)-(\d+)/(\d+)$")
46
47_SAFE_SEGMENT = re.compile(r"^[A-Za-z0-9._\-]+$")
48
49
50def safe_relpath(url_path: str) -> str | None:
51 raw = url_path.split("?", 1)[0].strip("/")
52 if not raw:
53 return None
54 segments = raw.split("/")
55 for seg in segments:
56 if seg in {"", ".", ".."} or not _SAFE_SEGMENT.match(seg):
57 return None
58 return "/".join(segments)
59
60
61def staging_path(rel: str) -> str:
62 sid = hashlib.sha256(rel.encode("utf-8")).hexdigest()
63 return os.path.join(STAGING_DIR, f"{sid}.upload")
64
65
66_locks_mu = threading.Lock()
67_locks: dict[str, threading.Lock] = {}
68
69
70def lock_for(rel: str) -> threading.Lock:
71 # Serializes concurrent POSTs targeting the same staging file. Without
72 # this, two threads handling the same chunk (e.g. a client retry after a
73 # lost response) can both pass the start==current_size check and both
74 # open("ab"), duplicating bytes and shifting the resume offset.
75 with _locks_mu:
76 lock = _locks.get(rel)
77 if lock is None:
78 lock = threading.Lock()
79 _locks[rel] = lock
80 return lock
81
82
83class Handler(http.server.BaseHTTPRequestHandler):
84 server_version = "ipfs-upload/1"
85
86 def do_POST(self) -> None: # noqa: N802 - stdlib interface
87 rel = safe_relpath(self.path)
88 if rel is None:
89 self.send_error(400, "invalid path")
90 return
91
92 try:
93 length = int(self.headers.get("Content-Length", "0"))
94 except ValueError:
95 self.send_error(400, "bad Content-Length")
96 return
97 if length < 0:
98 self.send_error(400, "bad Content-Length")
99 return
100
101 cr_header = self.headers.get("Content-Range")
102 staging = staging_path(rel)
103
104 with lock_for(rel):
105 self._handle_locked(rel, length, cr_header, staging)
106
107 def _handle_locked(
108 self, rel: str, length: int, cr_header: str | None, staging: str
109 ) -> None:
110 # Status check: empty body + no Content-Range.
111 if length == 0 and not cr_header:
112 try:
113 size = os.path.getsize(staging)
114 except OSError:
115 self.send_error(404, "no upload in progress")
116 return
117 body = (str(size) + "\n").encode("ascii")
118 self.send_response(200)
119 self.send_header("Content-Type", "text/plain; charset=utf-8")
120 self.send_header("Upload-Offset", str(size))
121 self.send_header("Content-Length", str(len(body)))
122 self.end_headers()
123 self.wfile.write(body)
124 return
125
126 # Parse Content-Range or derive a single-chunk request.
127 if cr_header:
128 m = _CONTENT_RANGE.match(cr_header)
129 if not m:
130 self.send_error(400, "bad Content-Range")
131 return
132 start, end, total = int(m.group(1)), int(m.group(2)), int(m.group(3))
133 if start < 0 or end < start or end >= total:
134 self.send_error(400, "bad Content-Range")
135 return
136 chunk_len = end - start + 1
137 if length != chunk_len:
138 self.send_error(400, "Content-Length doesn't match Content-Range")
139 return
140 else:
141 if length == 0:
142 self.send_error(411, "Content-Length required")
143 return
144 start, total, chunk_len = 0, length, length
145
146 # Current staging size and offset enforcement.
147 try:
148 current_size = os.path.getsize(staging)
149 except OSError:
150 current_size = 0
151
152 if start == 0:
153 mode = "wb" # truncate / fresh start
154 elif start == current_size:
155 mode = "ab"
156 else:
157 self.send_error(
158 409, f"offset mismatch: expected {current_size}, got {start}"
159 )
160 return
161
162 try:
163 with open(staging, mode) as f:
164 remaining = chunk_len
165 while remaining > 0:
166 data = self.rfile.read(min(CHUNK, remaining))
167 if not data:
168 break
169 f.write(data)
170 remaining -= len(data)
171 if remaining != 0:
172 raise OSError(f"short read: {remaining} bytes missing")
173 except OSError as e:
174 # Roll the staging file back to the chunk's start so the client
175 # can retry this chunk from a consistent offset.
176 try:
177 with open(staging, "r+b") as fix:
178 fix.truncate(start)
179 except OSError:
180 pass
181 self.send_error(500, f"write failed: {e}")
182 return
183
184 try:
185 new_size = os.path.getsize(staging)
186 except OSError as e:
187 self.send_error(500, f"stat failed: {e}")
188 return
189
190 if new_size < total:
191 self.send_response(308)
192 self.send_header("Upload-Offset", str(new_size))
193 self.send_header("Range", f"bytes=0-{new_size - 1}")
194 self.send_header("Content-Length", "0")
195 self.end_headers()
196 return
197
198 if new_size > total:
199 # Client over-sent (shouldn't happen given our checks, but guard).
200 try:
201 with open(staging, "r+b") as fix:
202 fix.truncate(total)
203 except OSError as e:
204 self.send_error(500, f"truncate failed: {e}")
205 return
206
207 # Complete: move staging → dst. The os.replace() below fires the
208 # ipfs-pin-watcher's inotify MOVED_TO, and the watcher is the SOLE
209 # pinner (`ipfs add --nocopy --pin`). We must NOT also pin here: two
210 # concurrent `--nocopy` adds of the same file corrupt the filestore
211 # into an incomplete DAG. So we only compute the CID to return to the
212 # client (--only-hash writes nothing). The watcher's single-threaded
213 # loop is then the only thing mutating the filestore. Pinning is
214 # therefore async — the watcher finishes a beat after this returns.
215 dst = os.path.join(DOWNLOAD_ROOT, rel)
216 dst_dir = os.path.dirname(dst) or DOWNLOAD_ROOT
217 try:
218 os.makedirs(dst_dir, exist_ok=True)
219 os.replace(staging, dst)
220 except OSError as e:
221 self.send_error(500, f"finalize failed: {e}")
222 return
223
224 try:
225 result = subprocess.run(
226 [IPFS_BIN, "add", "--only-hash", "--raw-leaves", "--quieter", dst],
227 check=True,
228 capture_output=True,
229 text=True,
230 )
231 except subprocess.CalledProcessError as e:
232 msg = (e.stderr or e.stdout or "").strip() or "unknown error"
233 self.send_error(500, f"ipfs add failed: {msg}")
234 return
235
236 cid = result.stdout.strip()
237 body = (cid + "\n").encode("ascii")
238 self.send_response(200)
239 self.send_header("Content-Type", "text/plain; charset=utf-8")
240 self.send_header("Content-Length", str(len(body)))
241 self.end_headers()
242 self.wfile.write(body)
243
244 def log_message(self, fmt: str, *args: object) -> None:
245 sys.stderr.write(f"{self.address_string()} {fmt % args}\n")
246
247
248class Server(socketserver.ThreadingMixIn, http.server.HTTPServer):
249 daemon_threads = True
250 allow_reuse_address = True
251
252
253def main() -> None:
254 os.makedirs(STAGING_DIR, exist_ok=True)
255 bind = os.environ.get("BIND", "127.0.0.1")
256 port = int(os.environ.get("PORT", "8090"))
257 with Server((bind, port), Handler) as s:
258 s.serve_forever()
259
260
261if __name__ == "__main__":
262 main()