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