tamakyi
8daa5b0bec
- serve.py：--host 默认改 0.0.0.0；启动时列出所有网卡地址（0.0.0.0 不能直接点开）； --host 127.0.0.1 仍是「只给本机」的收紧方式，README 说明公共网络用它。 - 改单线程为 ThreadingTCPServer：上传几 MB 的图会占住连接，单线程时 iframe 预览 和其它请求都得排队。 - 写接口加同源校验（Origin 与 Host 比对，无 Origin 放行）：挡住同网段里「另一个网页 借访客浏览器发写请求」，curl / 脚本不受影响。 - index.html：提示文案不再写死 127.0.0.1，改用 location.host —— 用局域网地址打开时 不该反过来提示「请用 127.0.0.1 打开」。 - mock_s3.py 加 --host（默认仍回环），便于在别的设备上联调。
185 lines
7.3 KiB
Python
185 lines
7.3 KiB
Python
"""最小 S3 兼容假服务，只为调试生成器的「测试连接 / 上传素材」。
|
||
|
||
它不校验签名（签名正确性由 check_sigv4.py 与官方 SDK 对拍保证），只校验
|
||
请求形态与权限分支，所以能用来演练：桶名写错、密钥写错、前缀为空、
|
||
虚拟主机风格这些配置。
|
||
|
||
独立跑：python dev/theme-builder/mock_s3.py --port 9900
|
||
被复用：from mock_s3 import MockS3, BUCKET
|
||
"""
|
||
import argparse
|
||
import sys
|
||
import threading
|
||
import urllib.parse
|
||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
|
||
BUCKET = "tb-assets"
|
||
|
||
|
||
class MockS3(BaseHTTPRequestHandler):
|
||
store = {}
|
||
requests = []
|
||
|
||
def log_message(self, fmt, *args):
|
||
pass
|
||
|
||
# ---- helpers ----
|
||
def _split(self):
|
||
path = urllib.parse.unquote(self.path.split("?")[0]).lstrip("/")
|
||
bucket, _, key = path.partition("/")
|
||
return bucket, key
|
||
|
||
def _key_id(self):
|
||
# Authorization 形如
|
||
# AWS4-HMAC-SHA256 Credential=AKID/20261021/us-east-1/s3/aws4_request, SignedHeaders=...
|
||
# 第一个逗号前既有算法名又有 Credential，不能按逗号切完再比前缀。
|
||
auth = self.headers.get("Authorization") or ""
|
||
i = auth.find("Credential=")
|
||
if i < 0:
|
||
return ""
|
||
return auth[i + len("Credential="):].split("/")[0]
|
||
|
||
def _body(self):
|
||
n = int(self.headers.get("Content-Length") or 0)
|
||
return self.rfile.read(n) if n else b""
|
||
|
||
def _xml(self, status, code, message):
|
||
body = ("<?xml version=\"1.0\"?><Error><Code>%s</Code><Message>%s</Message></Error>"
|
||
% (code, message)).encode()
|
||
self.send_response(status)
|
||
self.send_header("Content-Type", "application/xml")
|
||
if self.command != "HEAD":
|
||
# HEAD 不能有 body：真 S3 也只是返回状态行，客户端仍会按状态码报错。
|
||
self.send_header("Content-Length", str(len(body)))
|
||
self.end_headers()
|
||
if self.command != "HEAD":
|
||
self.wfile.write(body)
|
||
|
||
def _check(self, bucket):
|
||
MockS3.requests.append((self.command, self.path, dict(self.headers)))
|
||
if self._key_id() == "BADKEY":
|
||
self._xml(403, "SignatureDoesNotMatch",
|
||
"The request signature we calculated does not match")
|
||
return False
|
||
if bucket != BUCKET:
|
||
self._xml(404, "NoSuchBucket", "The specified bucket does not exist")
|
||
return False
|
||
if not (self.headers.get("x-amz-date") or "").endswith("Z"):
|
||
self._xml(403, "AccessDenied", "missing x-amz-date")
|
||
return False
|
||
return True
|
||
|
||
# ---- verbs ----
|
||
def do_HEAD(self):
|
||
bucket, _ = self._split()
|
||
if not self._check(bucket):
|
||
return
|
||
self.send_response(200)
|
||
self.send_header("Content-Length", "0")
|
||
self.end_headers()
|
||
|
||
def do_PUT(self):
|
||
bucket, key = self._split()
|
||
data = self._body()
|
||
if not self._check(bucket):
|
||
return
|
||
MockS3.store[key] = (data, self.headers.get("Content-Type"),
|
||
self.headers.get("Cache-Control"))
|
||
self.send_response(200)
|
||
self.send_header("ETag", '"mock"')
|
||
self.send_header("Content-Length", "0")
|
||
self.end_headers()
|
||
|
||
def do_GET(self):
|
||
parsed = urllib.parse.urlsplit(self.path)
|
||
qs = urllib.parse.parse_qs(parsed.query)
|
||
bucket, key = self._split()
|
||
if not key and qs.get("list-type") == ["2"]:
|
||
if not self._check(bucket):
|
||
return
|
||
return self._list(qs.get("prefix", [""])[0],
|
||
int(qs.get("max-keys", ["1000"])[0] or 1000),
|
||
int(qs.get("continuation-token", ["0"])[0] or 0))
|
||
MockS3.requests.append((self.command, self.path, dict(self.headers)))
|
||
if key not in MockS3.store:
|
||
self._xml(404, "NoSuchKey", "The specified key does not exist")
|
||
return
|
||
data, ctype, _ = MockS3.store[key]
|
||
self.send_response(200)
|
||
self.send_header("Content-Type", ctype or "application/octet-stream")
|
||
self.send_header("Content-Length", str(len(data)))
|
||
self.end_headers()
|
||
self.wfile.write(data)
|
||
|
||
def _list(self, prefix, max_keys, offset):
|
||
# 只实现 ListObjectsV2 里服务端用到的字段；分页用 offset 当 token，够测真实分页逻辑。
|
||
keys = sorted(k for k in MockS3.store if k.startswith(prefix))
|
||
page = keys[offset:offset + max_keys]
|
||
truncated = offset + len(page) < len(keys)
|
||
items = []
|
||
for k in page:
|
||
data = MockS3.store[k][0]
|
||
items.append(
|
||
"<Contents><Key>%s</Key><Size>%d</Size>"
|
||
"<LastModified>2026-10-21T00:00:00.000Z</LastModified><ETag>"mock"</ETag></Contents>"
|
||
% (k, len(data)))
|
||
token = ("<NextContinuationToken>%d</NextContinuationToken>" % (offset + len(page))) if truncated else ""
|
||
body = (
|
||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||
'<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">'
|
||
"<Name>%s</Name><Prefix>%s</Prefix><KeyCount>%d</KeyCount>"
|
||
"<MaxKeys>%d</MaxKeys><IsTruncated>%s</IsTruncated>%s%s"
|
||
"</ListBucketResult>"
|
||
% (BUCKET, prefix, len(page), max_keys, "true" if truncated else "false",
|
||
token, "".join(items))
|
||
).encode()
|
||
self.send_response(200)
|
||
self.send_header("Content-Type", "application/xml")
|
||
self.send_header("Content-Length", str(len(body)))
|
||
self.end_headers()
|
||
self.wfile.write(body)
|
||
|
||
def do_DELETE(self):
|
||
bucket, key = self._split()
|
||
if not self._check(bucket):
|
||
return
|
||
MockS3.store.pop(key, None)
|
||
self.send_response(204)
|
||
self.send_header("Content-Length", "0")
|
||
self.end_headers()
|
||
|
||
|
||
def spin(port=0, host="127.0.0.1"):
|
||
"""起在后台线程，返回 (server, endpoint)。port=0 让系统挑空闲端口。
|
||
|
||
host 默认只给本机用；如果生成器是在别的设备上打开的（局域网访问），
|
||
要把这里改成 0.0.0.0，并在面板里把 Endpoint 填成本机 IP。
|
||
"""
|
||
srv = ThreadingHTTPServer((host, port), MockS3)
|
||
threading.Thread(target=srv.serve_forever, daemon=True).start()
|
||
shown = "127.0.0.1" if host in ("0.0.0.0", "::") else host
|
||
return srv, "http://%s:%d" % (shown, srv.server_address[1])
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser(description="TamaBox 主题生成器用的假 S3")
|
||
ap.add_argument("--port", type=int, default=9900)
|
||
ap.add_argument("--host", default="127.0.0.1",
|
||
help="监听地址，默认 127.0.0.1；生成器在别的设备上打开时用 0.0.0.0")
|
||
args = ap.parse_args()
|
||
srv, endpoint = spin(args.port, args.host)
|
||
print("假 S3 已启动：%s" % endpoint)
|
||
print(" 桶名 %s，Access Key 随意（写 BADKEY 会返回 SignatureDoesNotMatch）" % BUCKET)
|
||
if args.host in ("0.0.0.0", "::"):
|
||
print(" 局域网设备访问时，面板里的 Endpoint 填 http://<本机IP>:%d" % args.port)
|
||
print("Ctrl+C 停止")
|
||
try:
|
||
threading.Event().wait()
|
||
except KeyboardInterrupt:
|
||
srv.shutdown()
|
||
print("\n已停止")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|