db47546c2b
3GOD was "three.js god" — a quick way to eyeball a GLB and drop it into a game. wardrobegod's stage, grid and publish flow cover that far better now, so the bench is retired. What survives is the half that quietly became infrastructure. Absorbed into depot/: server.py (271 lines, verbatim), its hardening tests, and viewer.html kept as superseded reference only. Retiring the REPO is safe; retiring the SERVICE is not. The live instance on the dealgod VPS holds 773 assets and two shipping games fetch from it at runtime — thriftgod web/index.html:1176 and procity loaders.js:7 both hardcode https://digalot.fyi/3god. So the /3god URL stays even though the name is gone: changing it means editing two shipping codebases for no benefit, and the name now survives only as a URL path nobody looks at. The VPS service is deployed, not sourced live from the Gitea checkout, so archiving that repo changes nothing at runtime. Checked before discarding the UI: the depot's /api/meta tag store had 0 tags and 0 notes across all 773 assets, so there was nothing to migrate. Tagging lives in library/index.json now. depot/README.md records the two things that are easy to get wrong and expensive to rediscover: publishing must go direct over the tailnet (the Cloudflare front 403s because auth trusts the raw socket peer — thriftgod's own prop_campaign.py --publish is broken for exactly this), and clean_name DELETES illegal characters so 'shop!-cat.glb' silently resolves to an existing different mesh. Verified after the change: both the public and tailnet endpoints still answer 200. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
272 lines
13 KiB
Python
272 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""3GOD — the prop house. One .glb depot + inspection room shared by THRIFTGOD,
|
|
90sDJsim and the robotmonster 3D store. Stdlib only, no DB — a folder of .glb files
|
|
plus a meta.json sidecar. Reads are public (the games load assets client-side in
|
|
every visitor's browser); writes need the password.
|
|
|
|
GOD3_PW=... python3 server.py [port] # default 8788
|
|
"""
|
|
import hashlib, hmac, ipaddress, json, os, re, socket, sys, threading, time, urllib.error, urllib.request
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from urllib.parse import unquote, urlparse
|
|
|
|
ROOT = os.path.dirname(os.path.abspath(__file__))
|
|
DIR = os.path.join(ROOT, 'assets')
|
|
THUMBS = os.path.join(DIR, '.thumbs') # viewer-captured 256px jpegs
|
|
os.makedirs(THUMBS, exist_ok=True)
|
|
META = os.path.join(DIR, 'meta.json')
|
|
OPEN = os.environ.get('GOD3_OPEN') == '1' # open-door testing: writes need no password
|
|
if OPEN:
|
|
print('🚪 GOD3_OPEN=1 — the prop house door is wide open (testing mode)')
|
|
PW = os.environ.get('GOD3_PW')
|
|
if not PW: # no shipped default — a locked depot on 'skeletonkey' is not locked
|
|
if not OPEN:
|
|
sys.exit('refusing to start: set GOD3_PW=<secret>, or GOD3_OPEN=1 for passwordless local testing.')
|
|
PW = 'skeletonkey' # OPEN bypasses auth anyway; token value is moot
|
|
TOKEN = hmac.new(PW.encode(), b'3god-ok', hashlib.sha256).hexdigest()
|
|
TS_ALLOW = {'100.89.131.57', '100.91.239.7', '100.102.229.52', '100.69.21.128', # workstations/laptops
|
|
'100.71.119.27', '100.94.195.115'} # VPS: botchat, dealgod
|
|
# tailnet peers: direct connection over tailscale = auto-write.
|
|
# ponytail: trusts the raw socket peer, NOT any X-Forwarded-For / CF-Connecting-IP header (spoofable). So it only grants
|
|
# when the machine hits this port DIRECTLY over the tailnet — through the digalot.fyi Cloudflare front the peer is CF, never 100.x.
|
|
MAX = 50 * 1024 * 1024
|
|
SAFE = re.compile(r'^[a-zA-Z0-9][a-zA-Z0-9._ -]*\.(glb|png|jpe?g|webp)$')
|
|
EXTS = ('.glb', '.png', '.jpg', '.jpeg', '.webp')
|
|
CT = {'.glb': 'model/gltf-binary', '.png': 'image/png', '.jpg': 'image/jpeg',
|
|
'.jpeg': 'image/jpeg', '.webp': 'image/webp'}
|
|
|
|
|
|
def looks_ok(data, name):
|
|
"""Magic bytes match the extension — no mystery files on the shelf."""
|
|
if name.endswith('.glb'):
|
|
return data[:4] == b'glTF'
|
|
if name.endswith('.png'):
|
|
return data[:8] == b'\x89PNG\r\n\x1a\n'
|
|
if name.endswith(('.jpg', '.jpeg')):
|
|
return data[:2] == b'\xff\xd8'
|
|
if name.endswith('.webp'):
|
|
return data[:4] == b'RIFF' and data[8:12] == b'WEBP'
|
|
return False
|
|
|
|
|
|
_META_LOCK = threading.Lock() # ThreadingHTTPServer → guard the load-mutate-save
|
|
|
|
|
|
def meta_load():
|
|
try:
|
|
return json.load(open(META))
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
def meta_save(m):
|
|
tmp = f'{META}.{os.getpid()}.tmp' # atomic: a torn write can never be read back as {}
|
|
with open(tmp, 'w') as f:
|
|
json.dump(m, f, indent=1)
|
|
os.replace(tmp, META)
|
|
|
|
|
|
def clean_name(n):
|
|
n = re.sub(r'[^a-zA-Z0-9._ -]', '', os.path.basename(n or '')).strip()
|
|
if n and not n.lower().endswith(EXTS):
|
|
n += '.glb'
|
|
return n if SAFE.match(n or '') else None
|
|
|
|
|
|
def cloud_url(u):
|
|
"""Paste-friendly: turn Drive/Dropbox share links into direct downloads."""
|
|
m = re.search(r'drive\.google\.com/file/d/([\w-]+)', u)
|
|
if m:
|
|
return f'https://drive.google.com/uc?export=download&id={m.group(1)}'
|
|
if 'dropbox.com' in u:
|
|
return re.sub(r'[?&]dl=0', '', u) + ('&' if '?' in u else '?') + 'dl=1'
|
|
return u
|
|
|
|
|
|
def public_host(url):
|
|
"""SSRF guard: True only for http(s) URLs whose host resolves to public IPs.
|
|
Blocks file://, ftp://, localhost, link-local (169.254.x metadata), and private ranges.
|
|
ponytail: does NOT close DNS-rebinding (re-resolve to internal after this check) — pin the
|
|
resolved IP and connect to it if 3GOD ever fetches on behalf of untrusted callers."""
|
|
p = urlparse(url)
|
|
if p.scheme not in ('http', 'https') or not p.hostname:
|
|
return False
|
|
try:
|
|
infos = socket.getaddrinfo(p.hostname, p.port or (443 if p.scheme == 'https' else 80),
|
|
proto=socket.IPPROTO_TCP)
|
|
except OSError:
|
|
return False
|
|
for *_, sockaddr in infos:
|
|
ip = ipaddress.ip_address(sockaddr[0])
|
|
if (ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved
|
|
or ip.is_multicast or ip.is_unspecified):
|
|
return False
|
|
return bool(infos)
|
|
|
|
|
|
class _GuardedRedirect(urllib.request.HTTPRedirectHandler):
|
|
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
|
if not public_host(newurl): # re-validate every redirect hop
|
|
raise urllib.error.HTTPError(newurl, code, 'redirect to non-public host blocked', headers, fp)
|
|
return super().redirect_request(req, fp, code, msg, headers, newurl)
|
|
|
|
|
|
_OPENER = urllib.request.build_opener(_GuardedRedirect) # http/https only (no file/ftp handlers), hops re-checked
|
|
|
|
|
|
class H(BaseHTTPRequestHandler):
|
|
def log_message(self, fmt, *a):
|
|
pass
|
|
|
|
def _send(self, code, body, ctype='application/json', extra=None):
|
|
data = body if isinstance(body, bytes) else json.dumps(body).encode()
|
|
self.send_response(code)
|
|
self.send_header('Content-Type', ctype)
|
|
self.send_header('Content-Length', len(data))
|
|
for k, v in (extra or {}).items():
|
|
self.send_header(k, v)
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
|
|
def authed(self):
|
|
if OPEN:
|
|
return True
|
|
if self.client_address[0] in TS_ALLOW: # trusted tailnet peer (direct connection only)
|
|
return True
|
|
c = self.headers.get('Cookie', '')
|
|
m = re.search(r'3g=([0-9a-f]{64})', c)
|
|
return bool(m and hmac.compare_digest(m.group(1), TOKEN))
|
|
|
|
def do_GET(self):
|
|
path = self.path.split('?')[0]
|
|
if path == '/' or path == '/index.html':
|
|
return self._send(200, open(os.path.join(ROOT, 'web', 'index.html'), 'rb').read(),
|
|
'text/html; charset=utf-8')
|
|
if path == '/api/list':
|
|
m = meta_load()
|
|
out = []
|
|
for f in sorted(os.listdir(DIR)):
|
|
if not f.endswith(EXTS):
|
|
continue
|
|
st = os.stat(os.path.join(DIR, f))
|
|
out.append(dict({'tags': [], 'note': ''}, **m.get(f, {}),
|
|
file=f, size=st.st_size, added=int(st.st_mtime),
|
|
thumb=os.path.isfile(os.path.join(THUMBS, f + '.jpg'))))
|
|
return self._send(200, {'assets': out, 'authed': self.authed()},
|
|
extra={'Access-Control-Allow-Origin': '*'})
|
|
if path.startswith('/t/'): # thumbnail (viewer-captured)
|
|
f = clean_name(unquote(path[3:]))
|
|
fp = f and os.path.join(THUMBS, f + '.jpg')
|
|
if not (fp and os.path.isfile(fp)):
|
|
return self._send(404, {'error': 'no thumb yet'})
|
|
return self._send(200, open(fp, 'rb').read(), 'image/jpeg',
|
|
{'Access-Control-Allow-Origin': '*', 'Cache-Control': 'public, max-age=3600'})
|
|
if path.startswith('/a/') or path.startswith('/dl/'):
|
|
f = clean_name(unquote(path.split('/', 2)[2]))
|
|
fp = f and os.path.join(DIR, f)
|
|
if not (fp and os.path.isfile(fp)):
|
|
return self._send(404, {'error': 'no such prop'})
|
|
extra = {'Access-Control-Allow-Origin': '*', 'Cache-Control': 'public, max-age=86400'}
|
|
if path.startswith('/dl/'):
|
|
extra['Content-Disposition'] = f'attachment; filename="{f}"'
|
|
return self._send(200, open(fp, 'rb').read(),
|
|
CT.get(os.path.splitext(f)[1], 'application/octet-stream'), extra)
|
|
return self._send(404, {'error': 'lost in the prop house'})
|
|
|
|
def do_POST(self):
|
|
path, _, q = self.path.partition('?')
|
|
n = int(self.headers.get('Content-Length') or 0)
|
|
if n > MAX:
|
|
return self._send(413, {'error': 'over 50MB — decimate it, digger'})
|
|
raw = self.rfile.read(n) if n else b''
|
|
|
|
if path == '/api/login':
|
|
body = json.loads(raw or '{}')
|
|
if hmac.compare_digest(str(body.get('password', '')), PW):
|
|
return self._send(200, {'ok': True},
|
|
extra={'Set-Cookie': f'3g={TOKEN}; Path=/; Max-Age=31536000; HttpOnly; SameSite=Lax'})
|
|
time.sleep(1) # gentle brake on guessing
|
|
return self._send(403, {'error': 'not the magic word'})
|
|
|
|
if not self.authed():
|
|
return self._send(403, {'error': 'password first'})
|
|
|
|
if path == '/api/upload': # raw body, ?name=file.glb|png|jpg|webp
|
|
name = clean_name(unquote(dict(p.split('=', 1) for p in q.split('&') if '=' in p).get('name', '')))
|
|
if not name:
|
|
return self._send(400, {'error': 'name it something.glb (or .png/.jpg/.webp)'})
|
|
if not looks_ok(raw, name):
|
|
return self._send(400, {'error': f'that is not really a {os.path.splitext(name)[1]} file'})
|
|
open(os.path.join(DIR, name), 'wb').write(raw)
|
|
return self._send(200, {'ok': True, 'file': name})
|
|
|
|
if path == '/api/thumb': # viewer snapshots the loaded model
|
|
name = clean_name(unquote(dict(p.split('=', 1) for p in q.split('&') if '=' in p).get('name', '')))
|
|
if not (name and os.path.isfile(os.path.join(DIR, name))):
|
|
return self._send(404, {'error': 'no such prop'})
|
|
if raw[:2] != b'\xff\xd8':
|
|
return self._send(400, {'error': 'thumbs are jpeg'})
|
|
open(os.path.join(THUMBS, name + '.jpg'), 'wb').write(raw)
|
|
return self._send(200, {'ok': True})
|
|
|
|
body = json.loads(raw or '{}')
|
|
|
|
if path == '/api/fetch': # paste a URL, server pulls it in
|
|
name = clean_name(body.get('name', ''))
|
|
if not name:
|
|
return self._send(400, {'error': 'name it something.glb'})
|
|
target = cloud_url(body.get('url', ''))
|
|
if not public_host(target):
|
|
return self._send(400, {'error': 'only public http(s) URLs allowed (no file://, localhost or internal IPs)'})
|
|
try:
|
|
req = urllib.request.Request(target, headers={'User-Agent': '3GOD/1.0'})
|
|
with _OPENER.open(req, timeout=60) as r:
|
|
data = r.read(MAX + 1)
|
|
except Exception as e:
|
|
return self._send(400, {'error': f'fetch failed: {e}'})
|
|
if len(data) > MAX:
|
|
return self._send(413, {'error': 'over 50MB'})
|
|
if not looks_ok(data, name):
|
|
return self._send(400, {'error': 'that URL is not the file type its name claims (got HTML? make the link public/direct)'})
|
|
open(os.path.join(DIR, name), 'wb').write(data)
|
|
return self._send(200, {'ok': True, 'file': name})
|
|
|
|
if path == '/api/del':
|
|
f = clean_name(body.get('file', ''))
|
|
fp = f and os.path.join(DIR, f)
|
|
if fp and os.path.isfile(fp):
|
|
os.remove(fp)
|
|
tp = os.path.join(THUMBS, f + '.jpg')
|
|
if os.path.isfile(tp):
|
|
os.remove(tp)
|
|
with _META_LOCK:
|
|
m = meta_load()
|
|
m.pop(f, None)
|
|
meta_save(m)
|
|
return self._send(200, {'ok': True})
|
|
|
|
if path == '/api/meta': # tags/notes + viewer-measured stats
|
|
f = clean_name(body.get('file', ''))
|
|
if not (f and os.path.isfile(os.path.join(DIR, f))):
|
|
return self._send(404, {'error': 'no such prop'})
|
|
with _META_LOCK:
|
|
m = meta_load()
|
|
row = m.setdefault(f, {})
|
|
for k in ('tags', 'note', 'clips', 'tris', 'dims', 'rot', 'scale'):
|
|
if k in body:
|
|
row[k] = body[k]
|
|
if isinstance(row.get('tags'), list): # cap + stringify (real XSS defense is escaping at render)
|
|
row['tags'] = [str(t)[:40] for t in row['tags'][:20]]
|
|
if 'note' in row:
|
|
row['note'] = str(row['note'])[:500]
|
|
meta_save(m)
|
|
return self._send(200, {'ok': True})
|
|
|
|
return self._send(404, {'error': 'lost in the prop house'})
|
|
|
|
|
|
if __name__ == '__main__':
|
|
port = int(sys.argv[1]) if len(sys.argv) > 1 else 8788
|
|
print(f'3GOD prop house on :{port} — {len([f for f in os.listdir(DIR) if f.endswith(EXTS)])} props on the shelf')
|
|
ThreadingHTTPServer(('0.0.0.0', port), H).serve_forever()
|