feat: 添加 OpenNexus 认证 Core 与 Stronghold 基础能力

server sync/sync_server/__main__.py

@@ -10,9 +10,18 @@ from .database import Database
 from .storage import S3Objects
 
 
+def application():
+    url = os.environ["SYNC_DATABASE_URL"]
+    if not url.startswith("postgresql+psycopg://"):
+        raise RuntimeError("生产入口只支持 PostgreSQL")
+    return create_app(Database(url), S3Objects(os.environ["SYNC_S3_ENDPOINT"], os.environ["SYNC_S3_BUCKET"]),
+                      Path(os.environ["SYNC_STAGING_DIR"]))
+
+
 def main():
     parser = argparse.ArgumentParser()
-    parser.add_argument("command", choices=["serve", "migrate", "create-user"])
+    parser.add_argument("command", choices=["serve", "migrate", "create-user", "cleanup-uploads"])
+    parser.add_argument("--workers", type=int, choices=[1, 2], default=2)
     parser.add_argument("--username")
     args = parser.parse_args()
     url = os.environ["SYNC_DATABASE_URL"]
@@ -24,8 +33,11 @@ def main():
         db.add_user(args.username or input("用户名: "), getpass.getpass("密码（至少12字符）: "))
     elif args.command == "serve":
         import uvicorn
-        app = create_app(db, S3Objects(os.environ["SYNC_S3_ENDPOINT"], os.environ["SYNC_S3_BUCKET"]), Path(os.environ["SYNC_STAGING_DIR"]))
-        uvicorn.run(app, host="0.0.0.0", port=8080, access_log=False)
+        uvicorn.run("sync_server.__main__:application", factory=True, workers=args.workers,
+                    host="0.0.0.0", port=8080, access_log=False)
+    elif args.command == "cleanup-uploads":
+        from .maintenance import cleanup_expired_uploads
+        print(cleanup_expired_uploads(db, Path(os.environ["SYNC_STAGING_DIR"])))
 
 
 if __name__ == "__main__":

server sync/sync_server/app.py

@@ -4,11 +4,17 @@ import hashlib
 import json
 import secrets
 import time
+import os
+import tempfile
+import asyncio
+from contextlib import asynccontextmanager
+import logging
 from pathlib import Path
 
 from fastapi import FastAPI, Header, Query, Request
 from fastapi.exceptions import RequestValidationError
-from fastapi.responses import JSONResponse, Response
+from fastapi.responses import JSONResponse, FileResponse
+from starlette.background import BackgroundTask
 
 from .database import Database, password_hash, row, rows, run
 from .models import Commit, Login, Refresh, Upload, VaultCreate
@@ -26,12 +32,36 @@ def digest(value: str) -> str:
 def create_app(db: Database, objects, staging: Path, *, quota=1024**3, clock=time.time):
     db.migrate()
     staging.mkdir(parents=True, exist_ok=True)
-    app = FastAPI(title="NotesAgent Sync", version="1.0.0")
+    @asynccontextmanager
+    async def lifespan(_app):
+        from .maintenance import cleanup_expired_uploads
+        stopping = asyncio.Event()
+
+        async def maintain():
+            while not stopping.is_set():
+                try:
+                    await asyncio.to_thread(cleanup_expired_uploads, db, staging, now=int(clock()))
+                except Exception:
+                    logging.getLogger(__name__).error("UPLOAD_MAINTENANCE_FAILED")
+                try:
+                    await asyncio.wait_for(stopping.wait(), timeout=60)
+                except TimeoutError:
+                    pass
+
+        worker = asyncio.create_task(maintain())
+        try:
+            yield
+        finally:
+            stopping.set()
+            await worker
+
+    app = FastAPI(title="OpenNexus Sync", version="1.0.0", lifespan=lifespan)
     app.state.database = db
 
     @app.exception_handler(SyncError)
     async def error(_request, exc):
-        return JSONResponse({"error": {"code": exc.code, "details": exc.details}}, status_code=exc.status)
+        headers = {"Retry-After": "60"} if exc.status == 429 else {}
+        return JSONResponse({"error": {"code": exc.code, "details": exc.details}}, status_code=exc.status, headers=headers)
 
     @app.exception_handler(RequestValidationError)
     async def invalid(_request, _exc):
@@ -66,12 +96,41 @@ def create_app(db: Database, objects, staging: Path, *, quota=1024**3, clock=tim
     def health():
         return {"status": "ok"}
 
-    @app.get("/ready")
-    def ready():
+    def readiness_probe():
         with db.transaction() as conn:
             if row(conn, "SELECT version FROM schema_version")["version"] != 1:
                 raise SyncError(503, "SCHEMA_INCOMPATIBLE")
-        return {"status": "ready", "schema": 1}
+        key = "health-probe/" + secrets.token_hex(16)
+        try:
+            with tempfile.TemporaryFile(dir=staging) as local:
+                local.write(b"opennexus-ready")
+                local.flush()
+                os.fsync(local.fileno())
+                local.seek(0)
+                if local.read() != b"opennexus-ready":
+                    raise OSError("STAGING_INTEGRITY")
+            objects.put(key, b"opennexus-ready")
+            if objects.get(key) != b"opennexus-ready":
+                raise OSError("STORAGE_INTEGRITY")
+        finally:
+            objects.delete(key)
+
+    ready_lock = asyncio.Lock()
+    ready_cache = {"until": 0.0, "ok": False}
+
+    @app.get("/ready")
+    async def ready():
+        async with ready_lock:
+            if time.monotonic() >= ready_cache["until"]:
+                try:
+                    await asyncio.wait_for(asyncio.to_thread(readiness_probe), timeout=3)
+                    ready_cache["ok"] = True
+                except Exception:
+                    ready_cache["ok"] = False
+                ready_cache["until"] = time.monotonic() + 5
+            if not ready_cache["ok"]:
+                raise SyncError(503, "DEPENDENCY_UNAVAILABLE")
+            return {"status": "ready", "schema": 1}
 
     @app.get("/sync/v1/handshake")
     def handshake(protocol: int = 1):
@@ -171,10 +230,26 @@ def create_app(db: Database, objects, staging: Path, *, quota=1024**3, clock=tim
             raise SyncError(404, "UPLOAD_EXPIRED")
         return upload
 
+    def reconcile_staging(upload):
+        path = staging / upload["id"]
+        try:
+            length = path.stat().st_size
+            if length < upload["offset_bytes"]:
+                raise SyncError(409, "UPLOAD_DAMAGED", {"restart_required": True})
+            if length > upload["offset_bytes"]:
+                with path.open("r+b") as stream:
+                    stream.truncate(upload["offset_bytes"])
+                    stream.flush()
+                    os.fsync(stream.fileno())
+        except OSError:
+            raise SyncError(409, "UPLOAD_DAMAGED", {"restart_required": True}) from None
+        return path
+
     @app.get("/sync/v1/vaults/{vault_id}/uploads/{upload_id}")
     def upload_status(vault_id: str, upload_id: str, authorization: str = Header(default="")):
         with db.transaction() as conn:
             upload = authorized_upload(conn, vault_id, upload_id, authorization)
+            reconcile_staging(upload)
             return {"offset": upload["offset_bytes"], "size": upload["size"], "expires": upload["expires"]}
 
     @app.put("/sync/v1/vaults/{vault_id}/uploads/{upload_id}")
@@ -187,6 +262,7 @@ def create_app(db: Database, objects, staging: Path, *, quota=1024**3, clock=tim
                 raise SyncError(413, "CHUNK_TOO_LARGE")
         with db.transaction() as conn:
             upload = authorized_upload(conn, vault_id, upload_id, authorization)
+            reconcile_staging(upload)
             if offset != upload["offset_bytes"]:
                 raise SyncError(409, "UPLOAD_OFFSET", {"offset": upload["offset_bytes"]})
             if offset + len(data) > upload["size"]:
@@ -212,15 +288,24 @@ def create_app(db: Database, objects, staging: Path, *, quota=1024**3, clock=tim
     @app.post("/sync/v1/vaults/{vault_id}/uploads/{upload_id}/complete")
     def complete_upload(vault_id: str, upload_id: str, authorization: str = Header(default="")):
         with db.transaction() as conn:
+            session, _ = vault(conn, vault_id, authorization, lock=True)
+            receipt = row(conn, "SELECT hash FROM upload_receipts WHERE id=:id AND vault_id=:v AND device_id=:d",
+                          id=upload_id, v=vault_id, d=session["device_id"])
+            if receipt:
+                return {"complete": True, "content_hash": receipt["hash"]}
             upload = authorized_upload(conn, vault_id, upload_id, authorization)
-            data = (staging / upload_id).read_bytes()
-            if len(data) != upload["size"] or len(data) != upload["offset_bytes"] or hashlib.sha256(data).hexdigest() != upload["hash"]:
+            path = reconcile_staging(upload)
+            with path.open("rb") as stream:
+                content_hash = hashlib.file_digest(stream, "sha256").hexdigest()
+            if upload["size"] != upload["offset_bytes"] or content_hash != upload["hash"]:
                 raise SyncError(422, "OBJECT_INTEGRITY")
             exists = row(conn, "SELECT hash FROM objects WHERE vault_id=:v AND hash=:h", v=vault_id, h=upload["hash"])
             if not exists:
-                objects.put(vault_id + "/" + upload["hash"], data)
-                run(conn, "INSERT INTO objects VALUES (:v,:h,:size,:now)", v=vault_id, h=upload["hash"], size=len(data), now=int(clock()))
-                run(conn, "UPDATE vaults SET used=used+:size WHERE id=:v", size=len(data), v=vault_id)
+                objects.put_file(vault_id + "/" + upload["hash"], path, upload["hash"])
+                run(conn, "INSERT INTO objects VALUES (:v,:h,:size,:now)", v=vault_id, h=upload["hash"], size=upload["size"], now=int(clock()))
+                run(conn, "UPDATE vaults SET used=used+:size WHERE id=:v", size=upload["size"], v=vault_id)
+            run(conn, "INSERT INTO upload_receipts VALUES (:id,:v,:d,:h,:now)", id=upload_id,
+                v=vault_id, d=session["device_id"], h=upload["hash"], now=int(clock()))
             run(conn, "DELETE FROM uploads WHERE id=:id", id=upload_id)
         (staging / upload_id).unlink(missing_ok=True)
         return {"complete": True, "content_hash": upload["hash"]}
@@ -288,9 +373,28 @@ def create_app(db: Database, objects, staging: Path, *, quota=1024**3, clock=tim
             obj = row(conn, "SELECT * FROM objects WHERE vault_id=:v AND hash=:h", v=vault_id, h=content_hash)
             if not obj:
                 raise SyncError(404, "OBJECT_NOT_FOUND")
-            data = objects.get(vault_id + "/" + content_hash)
-            if len(data) != obj["size"] or hashlib.sha256(data).hexdigest() != content_hash:
+            expected_size = obj["size"]
+        # Verify before returning any bytes, without holding a database transaction
+        # or buffering an entire attachment in RAM.
+        temporary = tempfile.NamedTemporaryFile(prefix="download-", dir=staging, delete=False)
+        path = Path(temporary.name)
+        try:
+            size, checksum = 0, hashlib.sha256()
+            with temporary, objects.open(vault_id + "/" + content_hash) as source:
+                while chunk := source.read(1048576):
+                    size += len(chunk)
+                    if size > expected_size:
+                        raise SyncError(503, "STORAGE_INTEGRITY")
+                    checksum.update(chunk)
+                    temporary.write(chunk)
+            if size != expected_size or checksum.hexdigest() != content_hash:
                 raise SyncError(503, "STORAGE_INTEGRITY")
-            return Response(data, media_type="application/octet-stream", headers={"ETag": '"' + content_hash + '"', "Cache-Control": "private, no-store"})
+            return FileResponse(path, media_type="application/octet-stream",
+                headers={"ETag": '"' + content_hash + '"', "Cache-Control": "private, no-store"},
+                background=BackgroundTask(path.unlink, missing_ok=True))
+        except BaseException:
+            temporary.close()
+            path.unlink(missing_ok=True)
+            raise
 
     return app

server sync/sync_server/database.py

@@ -16,6 +16,7 @@ SCHEMA = [
     "CREATE TABLE IF NOT EXISTS revisions (vault_id TEXT NOT NULL, sequence BIGINT NOT NULL, file_id TEXT NOT NULL, base_revision BIGINT NOT NULL, path TEXT NOT NULL, path_key TEXT NOT NULL, operation TEXT NOT NULL, hash TEXT, size BIGINT NOT NULL, device_id TEXT NOT NULL, operation_id TEXT NOT NULL, fingerprint TEXT NOT NULL, PRIMARY KEY(vault_id, sequence), UNIQUE(vault_id, operation_id))",
     "CREATE TABLE IF NOT EXISTS files (vault_id TEXT NOT NULL, file_id TEXT NOT NULL, sequence BIGINT NOT NULL, path_key TEXT NOT NULL, deleted INTEGER NOT NULL, PRIMARY KEY(vault_id, file_id))",
     "CREATE TABLE IF NOT EXISTS login_limits (key TEXT PRIMARY KEY, started BIGINT NOT NULL, attempts INTEGER NOT NULL)",
+    "CREATE TABLE IF NOT EXISTS upload_receipts (id TEXT PRIMARY KEY, vault_id TEXT NOT NULL, device_id TEXT NOT NULL, hash TEXT NOT NULL, completed BIGINT NOT NULL)",
 ]
 
 
@@ -27,11 +28,16 @@ def password_hash(password: str, salt: str | None = None) -> str:
 
 class Database:
     def __init__(self, url: str):
-        self.engine = create_engine(url)
+        options = {"connect_args": {"connect_timeout": 2, "options": "-c statement_timeout=30000 -c lock_timeout=5000"},
+                   "pool_timeout": 2, "pool_pre_ping": True} if url.startswith("postgresql") else {}
+        self.engine = create_engine(url, **options)
         self.sqlite = self.engine.dialect.name == "sqlite"
 
     def migrate(self):
         with self.transaction() as conn:
+            if not self.sqlite:
+                # Serialize factory startup migrations across the supported workers.
+                conn.execute(text("SELECT pg_advisory_xact_lock(1330534488)"))
             conn.execute(text(SCHEMA[0]))
             version = conn.execute(text("SELECT version FROM schema_version")).scalar()
             if version not in {None, 1}:

server sync/sync_server/maintenance.py

@@ -0,0 +1,26 @@
+"""Delete expired upload staging only; referenced historical objects are never GC'd."""
+import re
+import time
+from pathlib import Path
+from .database import row, rows, run
+
+
+def cleanup_expired_uploads(db, staging: Path, *, now=None, limit=500):
+    now = int(time.time() if now is None else now)
+    with db.transaction() as conn:
+        expired = rows(conn, "SELECT id,vault_id FROM uploads WHERE expires<=:now ORDER BY expires LIMIT :limit", now=now, limit=limit)
+    removed = 0
+    for candidate in expired:
+        # Same lock order as PUT/complete. Recheck expiry after acquiring the lock.
+        with db.transaction() as conn:
+            suffix = " FOR UPDATE" if not db.sqlite else ""
+            row(conn, "SELECT id FROM vaults WHERE id=:v" + suffix, v=candidate["vault_id"])
+            upload = row(conn, "SELECT id FROM uploads WHERE id=:id AND expires<=:now", id=candidate["id"], now=now)
+            if upload:
+                if not re.fullmatch(r"[0-9a-f]{32}", upload["id"]):
+                    raise RuntimeError("UPLOAD_ID_INVALID")
+                # File first: interruption leaves an expired row that can be retried.
+                (staging / upload["id"]).unlink(missing_ok=True)
+                run(conn, "DELETE FROM uploads WHERE id=:id", id=upload["id"])
+                removed += 1
+    return {"expired_uploads_removed": removed}

server sync/sync_server/storage.py

@@ -4,6 +4,8 @@ from pathlib import Path
 import hashlib
 import os
 import tempfile
+import shutil
+from contextlib import contextmanager
 
 
 class DiskObjects:
@@ -27,6 +29,30 @@ class DiskObjects:
     def get(self, key: str) -> bytes:
         return (self.root / key).read_bytes()
 
+    def put_file(self, key: str, path: Path, content_hash: str):
+        target = self.root / key
+        target.parent.mkdir(parents=True, exist_ok=True)
+        with tempfile.NamedTemporaryFile(dir=target.parent, delete=False) as stream:
+            temporary = Path(stream.name)
+            try:
+                with path.open("rb") as source:
+                    shutil.copyfileobj(source, stream, 1024 * 1024)
+                stream.flush()
+                os.fsync(stream.fileno())
+            except BaseException:
+                stream.close()
+                temporary.unlink(missing_ok=True)
+                raise
+        try:
+            os.replace(temporary, target)
+        finally:
+            temporary.unlink(missing_ok=True)
+
+    @contextmanager
+    def open(self, key: str):
+        with (self.root / key).open("rb") as stream:
+            yield stream
+
     def delete(self, key: str):
         (self.root / key).unlink(missing_ok=True)
 
@@ -34,7 +60,10 @@ class DiskObjects:
 class S3Objects:
     def __init__(self, endpoint: str, bucket: str):
         import boto3
-        self.client = boto3.client("s3", endpoint_url=endpoint)
+        from botocore.config import Config
+        self.client = boto3.client("s3", endpoint_url=endpoint,
+                                   config=Config(connect_timeout=2, read_timeout=2,
+                                                 retries={"max_attempts": 0}))
         self.bucket = bucket
 
     def put(self, key: str, data: bytes):
@@ -46,5 +75,18 @@ class S3Objects:
         with response["Body"] as stream:
             return stream.read()
 
+    def put_file(self, key: str, path: Path, content_hash: str):
+        from boto3.s3.transfer import TransferConfig
+        with path.open("rb") as stream:
+            self.client.upload_fileobj(stream, self.bucket, key,
+                ExtraArgs={"Metadata": {"sha256": content_hash}},
+                Config=TransferConfig(use_threads=False, max_concurrency=1))
+
+    @contextmanager
+    def open(self, key: str):
+        response = self.client.get_object(Bucket=self.bucket, Key=key)
+        with response["Body"] as stream:
+            yield stream
+
     def delete(self, key: str):
         self.client.delete_object(Bucket=self.bucket, Key=key)

server sync/tests/test_production_storage.py

@@ -0,0 +1,99 @@
+"""Fault vectors for the production IO paths; database still uses an isolated fixture."""
+import hashlib
+import pytest
+from fastapi.testclient import TestClient
+from sync_server.app import create_app
+from sync_server.database import Database, row
+from sync_server.maintenance import cleanup_expired_uploads
+from sync_server.storage import DiskObjects
+from test_protocol import setup, upload
+
+
+@pytest.fixture
+def env(tmp_path):
+    db = Database("sqlite:///" + str(tmp_path / "sync.db"))
+    store = DiskObjects(tmp_path / "objects")
+    staging = tmp_path / "staging"
+    now = [1000]
+    app = create_app(db, store, staging, clock=lambda: now[0])
+    db.add_user("alice", "controlled-fixture-password")
+    with TestClient(app) as client:
+        yield client, db, store, staging, now
+    db.engine.dispose()
+
+
+def test_offset_reconciliation_never_acknowledges_missing_disk_bytes(env):
+    client, _, _, staging, _ = env
+    auth, _, base = setup(client)
+    info = client.post(base + "/uploads", headers=auth,
+                       json={"content_hash": hashlib.sha256(b"abc").hexdigest(), "size": 3}).json()
+    path = base + "/uploads/" + info["upload_id"]
+    assert client.put(path + "?offset=0", headers=auth, content=b"a").status_code == 200
+    local = staging / info["upload_id"]
+    local.write_bytes(b"abc")
+    assert client.get(path, headers=auth).json()["offset"] == 1
+    assert local.read_bytes() == b"a"
+    local.write_bytes(b"")
+    assert client.get(path, headers=auth).json()["error"]["code"] == "UPLOAD_DAMAGED"
+    assert client.put(path + "?offset=1", headers=auth, content=b"bc").status_code == 409
+
+
+def test_complete_retry_has_durable_receipt_and_charges_once(env):
+    client, db, _, staging, now = env
+    auth, _, base = setup(client)
+    info = client.post(base + "/uploads", headers=auth,
+                       json={"content_hash": hashlib.sha256(b"abc").hexdigest(), "size": 3}).json()
+    path = base + "/uploads/" + info["upload_id"]
+    assert client.put(path + "?offset=0", headers=auth, content=b"abc").status_code == 200
+    first = client.post(path + "/complete", headers=auth)
+    assert first.status_code == 200
+    assert not (staging / info["upload_id"]).exists()
+    for _ in range(100):
+        retry = client.post(path + "/complete", headers=auth)
+        assert retry.status_code == 200
+        assert retry.json() == first.json()
+    assert client.post(path + "/complete").status_code == 401
+    with db.transaction() as conn:
+        assert row(conn, "SELECT used FROM vaults WHERE id=:id", id=base.rsplit("/", 1)[1])["used"] == 3
+        assert row(conn, "SELECT COUNT(*) AS n FROM upload_receipts")["n"] == 1
+
+
+def test_download_is_verified_before_response_and_temp_files_are_removed(env):
+    client, _, store, staging, _ = env
+    auth, _, base = setup(client)
+    sha = upload(client, base, auth)
+    # This path must use streaming open(), never the full-object get().
+    store.get = lambda key: pytest.fail("full-object read")
+    response = client.get(base + "/objects/" + sha, headers=auth)
+    assert response.content == b"controlled note"
+    assert list(staging.glob("download-*")) == []
+    (store.root / base.rsplit("/", 1)[1] / sha).write_bytes(b"corruption")
+    response = client.get(base + "/objects/" + sha, headers=auth)
+    assert response.status_code == 503
+    assert b"corruption" not in response.content
+    assert list(staging.glob("download-*")) == []
+
+
+def test_cleanup_only_expired_staging_preserves_committed_objects(env):
+    client, db, store, staging, now = env
+    auth, _, base = setup(client)
+    sha = upload(client, base, auth)
+    info = client.post(base + "/uploads", headers=auth,
+                       json={"content_hash": hashlib.sha256(b"pending").hexdigest(), "size": 7}).json()
+    assert cleanup_expired_uploads(db, staging, now=1001)["expired_uploads_removed"] == 0
+    now[0] += 3601
+    assert cleanup_expired_uploads(db, staging, now=now[0])["expired_uploads_removed"] == 1
+    assert not (staging / info["upload_id"]).exists()
+    assert cleanup_expired_uploads(db, staging, now=now[0])["expired_uploads_removed"] == 0
+    assert store.get(base.rsplit("/",1)[1] + "/" + sha) == b"controlled note"
+    with db.transaction() as conn:
+        assert row(conn,"SELECT used FROM vaults")["used"] == len(b"controlled note")
+
+
+def test_readiness_fails_closed_when_object_storage_unavailable(env):
+    client, _, store, _, _ = env
+    def unavailable(*args):
+        raise OSError("simulated outage")
+    store.put = unavailable
+    assert client.get("/ready").status_code == 503
+    assert client.get("/health").status_code == 200