From dfb4fe4e1aaaf6d72eafcecf25777bb2f033b90b Mon Sep 17 00:00:00 2001 From: TolaMironcenko Date: Sat, 31 Oct 2026 23:30:12 +0700 Subject: [PATCH] Implement ecosystem contract v2: capabilities, shared ledgers, attachments MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - GET /v1/capabilities and the extended sync entity allowlist (§0, §1). - Shared ledgers (§7): ledgers, members and hashed one-time invite codes; X-Balance-Ledger selects the space on /v1/sync and /v1/attachments with its own records, change stream and cursor. Personal space keeps the v1 tables and behaviour; the migration only adds tables. - Attachments (§8): SHA-256, size and media type checks, blob files in BALANCE_ATTACHMENTS_DIR (default: attachments next to the database), garbage collection of unreferenced blobs with a grace period for fresh uploads, and ledger deletion removes its blobs. - Tests for every new endpoint and error code; API, sync, data model, security, deployment and README docs updated. Co-Authored-By: Claude Opus 5.5 --- AGENTS.md | 3 +- README.md | 16 +- docs/API.md | 77 +++- docs/ARCHITECTURE.md | 2 +- docs/CONTRACT_V2.md | 258 ++++++++++++ docs/DATA_MODEL.md | 34 +- docs/DEPLOYMENT.md | 1 + docs/SECURITY.md | 19 + docs/SYNC.md | 29 ++ src/attachments.rs | 326 +++++++++++++++ src/config.rs | 33 +- src/database.rs | 79 +++- src/ledgers.rs | 365 +++++++++++++++++ src/main.rs | 2 + src/server.rs | 325 +++++++++++++-- src/server/v2_tests.rs | 892 +++++++++++++++++++++++++++++++++++++++++ src/sync.rs | 91 ++++- 17 files changed, 2496 insertions(+), 56 deletions(-) create mode 100644 docs/CONTRACT_V2.md create mode 100644 src/attachments.rs create mode 100644 src/ledgers.rs create mode 100644 src/server/v2_tests.rs diff --git a/AGENTS.md b/AGENTS.md index 17def6b..36307cf 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -14,7 +14,8 @@ Go — в `tests/fixtures/go_schema_v1.sql`. Оба файла не правь вручную. Основные Rust-модули: `src/main.rs`, `runtime.rs`, `config.rs`, -`database.rs`, `auth.rs`, `server.rs`, `sync.rs`, `go_json.rs`. +`database.rs`, `auth.rs`, `server.rs`, `sync.rs`, `go_json.rs`, +`ledgers.rs`, `attachments.rs` (контракт v2, `docs/CONTRACT_V2.md`). Веб-клиент из `web/` встроен в Rust executable. Обнаруженные технологии/интеграции: SQLite, Docker. diff --git a/README.md b/README.md index 934b922..0cdcc05 100644 --- a/README.md +++ b/README.md @@ -76,6 +76,7 @@ The browser session uses `HttpOnly`, `SameSite=Strict` cookies. Native applicati | `BALANCE_ACCESS_TTL` | `15m` | Access-token lifetime | | `BALANCE_REFRESH_TTL` | `720h` | Refresh-token lifetime | | `BALANCE_CORS_ORIGINS` | empty | Comma-separated browser origins; native apps do not need CORS | +| `BALANCE_ATTACHMENTS_DIR` | `attachments` next to `BALANCE_DB_PATH` | Directory for attachment files (receipt photos); Docker: `/data/attachments` | ## HTTPS @@ -99,15 +100,24 @@ Then enter `https://balance.example.com` in each application's Settings → Serv | `POST` | `/v1/auth/refresh` | Rotate session tokens | | `POST` | `/v1/auth/logout` | Revoke refresh token | | `GET` | `/v1/me` | Current account | -| `POST` | `/v1/sync` | Push and pull changes | +| `POST` | `/v1/sync` | Push and pull changes (`X-Balance-Ledger` selects a shared ledger) | +| `GET` | `/v1/capabilities` | Contract version and features (no authorization) | +| `GET`, `POST` | `/v1/ledgers` | List / create shared ledgers | +| `PATCH`, `DELETE` | `/v1/ledgers/{id}` | Rename / delete a ledger (owner) | +| `POST` | `/v1/ledgers/{id}/invites` | One-time invite code (owner) | +| `POST` | `/v1/ledgers/join` | Join a ledger by code | +| `DELETE` | `/v1/ledgers/{id}/members/{userId}` | Remove a member or leave | +| `PUT`, `GET` | `/v1/attachments/{sha256}` | Upload / download attachment bytes | -The sync endpoint accepts transactions, custom categories and monthly budgets. The server isolates records by account, stores an append-only change cursor, and resolves competing edits with last-write-wins timestamps. Current Apple and Android clients use push-only requests followed by paginated pulls; update the server and applications together. +The server implements ecosystem contract v2 ([docs/CONTRACT_V2.md](docs/CONTRACT_V2.md)). +The sync endpoint accepts transactions, categories, budgets, accounts, transfers, +recurring operations, goals, debts, exchange rates, settings and attachment metadata. The server isolates records by account, stores an append-only change cursor, and resolves competing edits with last-write-wins timestamps. Current Apple and Android clients use push-only requests followed by paginated pulls; update the server and applications together. ## Operational notes - Registration can be disabled after the required accounts have been created. - Passwords are stored as salted Argon2id hashes. Refresh tokens are stored only as SHA-256 hashes. -- Back up the SQLite database and its `-wal`/`-shm` files together, or stop the server before copying the database. +- Back up the SQLite database and its `-wal`/`-shm` files together with the attachments directory, or stop the server before copying them. - Changing `BALANCE_JWT_SECRET` invalidates access tokens. Existing refresh sessions remain valid and will receive new access tokens after refresh. - The in-process authentication limiter allows 30 authentication requests per IP per minute. Use a reverse proxy for stronger public rate limiting and request logging. diff --git a/docs/API.md b/docs/API.md index d32bddc..dcb41d9 100644 --- a/docs/API.md +++ b/docs/API.md @@ -82,14 +82,21 @@ POST /v1/sync } ``` -Поддерживаемые entity: +Поддерживаемые entity (контракт v2, `docs/CONTRACT_V2.md` §1): ```text -transaction -category -budget +transaction category budget account transfer recurring +goal debt rate setting attachment ``` +Незнакомая entity отклоняется `400 invalid_change`; текст сообщения +(`entity must be transaction, category, or budget`) сохранён из v1, потому что +он записан в эталоне `tools/compatibility_expected.txt`. Payload сервер хранит +как есть и не валидирует по сущности. + +Заголовок `X-Balance-Ledger: ` выбирает общее пространство (см. ниже). Без +заголовка (или с пустым значением) — личное пространство, как в v1. + ## Push Client отправляет локальные изменения. @@ -112,6 +119,68 @@ Server возвращает изменения после этого cursor, в Если существует pagination, ответ должен содержать понятный признак продолжения и новый cursor/sequence. +## Контракт v2 + +### Capabilities + +`GET /v1/capabilities` — без авторизации: + +```json +{"contract":2, + "entities":["transaction","category","budget","account","transfer","recurring","goal","debt","rate","setting","attachment"], + "attachments":{"maxBytes":10485760,"types":["image/jpeg","image/png","image/heic","application/pdf"]}, + "ledgers":true} +``` + +### Общие пространства (ledgers) + +Все запросы — с `Authorization`. Объект ledger: +`{"id","name","role","members":[{"userId","email","role"}]}`, `role` — +роль текущего пользователя (`owner` | `member`), владелец идёт первым в `members`. + +| Метод | Путь | Успех | Кто | +| --- | --- | --- | --- | +| GET | `/v1/ledgers` | `200 {"ledgers":[…]}` (только общие) | любой | +| POST | `/v1/ledgers` `{"name"}` | `201` ledger, `role:"owner"` | любой | +| PATCH | `/v1/ledgers/{id}` `{"name"}` | `200` ledger | владелец | +| DELETE | `/v1/ledgers/{id}` | `204` | владелец; удаляет записи, курсор, приглашения, участников и файлы вложений | +| POST | `/v1/ledgers/{id}/invites` | `201 {"code","expiresAt"}` | владелец | +| POST | `/v1/ledgers/join` `{"code"}` | `200` ledger, `role:"member"` | любой | +| DELETE | `/v1/ledgers/{id}/members/{userId}` | `204` | владелец — любого участника; участник — себя | + +- Имя: после `trim` 1–100 символов без управляющих → иначе `400 invalid_name`. +- Код приглашения: 10 символов из `ABCDEFGHJKLMNPQRSTUVWXYZ23456789`, живёт 7 дней, + одноразовый. При вводе регистр, пробелы и `-` игнорируются. Повторный join + участника возвращает ledger и код не расходует. `/v1/ledgers/join` входит в + общий лимит 30 запросов/мин на IP вместе с auth endpoints. +- Ошибки: `400 invalid_code` (неверный, использованный или просроченный код), + `403 forbidden` (не участник или не владелец), `404 not_found` (нет такого + пространства или участника), `409 owner_cannot_leave` (владелец пытается + выйти — он может только удалить пространство), `400 invalid_json`. +- `X-Balance-Ledger` на `/v1/sync` и `/v1/attachments/*`: не участник → `403 + forbidden`, несуществующее → `404 not_found`. + +### Вложения + +| Метод | Путь | Ответ | +| --- | --- | --- | +| PUT | `/v1/attachments/{sha256}` | тело — байты; `201` первый раз, `200` повтор; `{"sha256","size","contentType"}` | +| GET/HEAD | `/v1/attachments/{sha256}` | `200`, байты, сохранённый `Content-Type`, `Content-Disposition: attachment`; `404 not_found`, если в пространстве нет файла | + +- `{sha256}` — 64 hex-символа в нижнем регистре; SHA-256 тела должен совпасть, + иначе `400 hash_mismatch` (так же для неверного формата хеша в PUT). +- `Content-Type` (параметры игнорируются) — `image/jpeg`, `image/png`, + `image/heic`, `application/pdf`, иначе `415 unsupported_media_type`. +- Больше 10 MiB → `413 payload_too_large`. Лимит тела 10 MiB действует только для + авторизованного PUT; остальные запросы по-прежнему ограничены 2 MiB. + +### CORS + +Ответы v1 сохраняют заголовки эталона Go. Для маршрутов v2 и для настоящих +preflight-запросов (есть `Access-Control-Request-Method`) разрешённые origin +получают `Access-Control-Allow-Methods: GET, POST, PUT, PATCH, DELETE, OPTIONS` +и `Access-Control-Allow-Headers: Authorization, Content-Type, X-Balance-Ledger`. + ## Ошибки Рекомендуемый контракт: diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 771ea95..691120b 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -23,7 +23,7 @@ Android ───────────┘ ## Исходный код -Production runtime — один Rust-процесс: `src/main.rs`, `runtime.rs`, `config.rs`, `database.rs`, `auth.rs`, `sync.rs`, `server.rs`, `go_json.rs`. SQLite bundled через rusqlite; HTTP обслуживается axum/hyper. `runtime.rs` сохраняет socket deadlines Go (header 5s, read 15s, write 30s, idle 60s) и ограничивает graceful shutdown десятью секундами. `go_json.rs` сохраняет особенности encoding/json и исходные payload bytes. Статические файлы из `web/` встроены при компиляции. +Production runtime — один Rust-процесс: `src/main.rs`, `runtime.rs`, `config.rs`, `database.rs`, `auth.rs`, `sync.rs`, `server.rs`, `go_json.rs`, `ledgers.rs` (общие пространства контракта v2), `attachments.rs` (файлы вложений на диске и их сборка мусора). SQLite bundled через rusqlite; HTTP обслуживается axum/hyper. `runtime.rs` сохраняет socket deadlines Go (header 5s, read 15s, write 30s, idle 60s) и ограничивает graceful shutdown десятью секундами. `go_json.rs` сохраняет особенности encoding/json и исходные payload bytes. Статические файлы из `web/` встроены при компиляции. Маршруты контракта v2 (`/v1/capabilities`, `/v1/ledgers*`, `/v1/attachments/*`) разбираются в `server.rs` до маршрутов v1, которые сохраняют поведение эталона Go. Исходная Go-реализация удалена после v1.0.0 (исходники остались в теге `v1.0.0`). Её наблюдаемое поведение зафиксировано в `tools/compatibility_expected.txt`, а SQLite-схема — в `tests/fixtures/go_schema_v1.sql`; Rust сверяется с обоими в CI. diff --git a/docs/CONTRACT_V2.md b/docs/CONTRACT_V2.md new file mode 100644 index 0000000..a899188 --- /dev/null +++ b/docs/CONTRACT_V2.md @@ -0,0 +1,258 @@ +# Контракт экосистемы v2 + +> Общий документ BalanceServer, BalanceFlutter и BalanceWatch. Одинаковая +> копия лежит в каждом репозитории: `docs/ecosystem/CONTRACT_V2.md` +> (Flutter), `docs/CONTRACT_V2.md` (сервер, часы). Меняется только во всех +> трёх сразу. + +v2 добавляет счета и переводы, регулярные операции, мультивалютность, цели, +теги, вложения (фото чеков), общие бюджеты и долги. Всё сделано +**добавлением**: формат `/v1/sync`, курсор, `sequence`, `updatedAt`, +tombstone и last-write-wins не меняются. Клиент v1 продолжает работать с +сервером v2, но не видит новых данных (ограничения — в конце). + +## 0. Общие правила + +1. **Незнакомая сущность** при pull пропускается молча (курсор двигается). +2. **Незнакомые поля** payload знакомой сущности клиент v2 **сохраняет** и + отправляет обратно без изменений при следующем сохранении записи + (forward compatibility). Клиент v1 их теряет — поэтому обновлять нужно все + клиенты. +3. Все новые поля payload необязательные. Отсутствие поля = значение по + умолчанию, указанное ниже. +4. Деньги на проводе — числа (`amount`, `limit`, `target`…), как в v1; + внутри клиентов — целые сотые. Курс валют — **строка** десятичного числа. +5. Даты — ISO 8601 UTC. Интервалы — `[начало, конец)`. +6. Коды валют — ISO 4217 в верхнем регистре (`RUB`, `EUR`, `USD`, `SEK`, …). +7. Идентификаторы записей — UUID в нижнем регистре, кроме `setting` (ключ). + +### Возможности сервера + +`GET /v1/capabilities` (без авторизации): + +```json +{ + "contract": 2, + "entities": ["transaction","category","budget","account","transfer", + "recurring","goal","debt","rate","setting","attachment"], + "attachments": {"maxBytes": 10485760, + "types": ["image/jpeg","image/png","image/heic","application/pdf"]}, + "ledgers": true +} +``` + +Сервер v1 отвечает 404 — клиент v2 тогда скрывает вложения и общие бюджеты и +не отправляет новые сущности. + +## 1. Сущности + +Белый список сервера: `transaction`, `category`, `budget`, `account`, +`transfer`, `recurring`, `goal`, `debt`, `rate`, `setting`, `attachment`. + +### transaction — новые поля + +| Поле | Тип | По умолчанию | Смысл | +| --- | --- | --- | --- | +| `accountId` | string? | нет | счёт; без него — «Без счёта» | +| `currency` | string? | валюта счёта, иначе базовая | валюта суммы | +| `tags` | [string]? | `[]` | теги, см. §6 | +| `recurringId` | string? | нет | создана из регулярной операции | +| `occurrence` | string? | нет | дата повторения `YYYY-MM-DD` | + +`kindRawValue` остаётся `income` / `expense`. Перевод — **не** транзакция. + +### account + +```json +{"name":"Карта","kindRawValue":"card","icon":"creditcard.fill", + "colorName":"blue","currency":"RUB","createdAt":"…","isArchived":false, + "sortOrder":0} +``` + +`kindRawValue`: `cash` | `card` | `deposit` | `other` (незнакомое → `other`). +`currency` по умолчанию — базовая. + +### transfer + +```json +{"fromAccountId":"…","toAccountId":"…","amount":1000.0,"toAmount":10.5, + "date":"…","note":""} +``` + +`toAmount` нужен, когда валюты счетов различаются; по умолчанию = `amount`. +`amount` — в валюте счёта-источника, `toAmount` — в валюте счёта-получателя. + +### recurring + +```json +{"kindRawValue":"expense","amount":990.0,"currency":"RUB","accountId":null, + "categoryName":"Подписки","categoryIcon":"repeat","categoryEmoji":"", + "categoryColorName":"cyan","note":"Связь","tags":[], + "frequency":"monthly","interval":1,"startDate":"2026-01-05T09:00:00Z", + "endDate":null,"mode":"auto","isPaused":false,"generatedThrough":null} +``` + +`frequency`: `daily` | `weekly` | `monthly` | `yearly`; `interval` ≥ 1; +`mode`: `auto` (создаётся само) | `remind` (напоминание, пользователь +подтверждает). `generatedThrough` — дата `YYYY-MM-DD` последнего +обработанного повторения. + +### goal + +```json +{"name":"Отпуск","target":150000.0,"currency":"RUB","deadline":null, + "accountId":null,"saved":20000.0,"icon":"airplane","colorName":"sky", + "createdAt":"…","isArchived":false} +``` + +Прогресс = баланс счёта `accountId`, если он задан, иначе `saved`. + +### debt + +```json +{"person":"Иван","direction":"owedToMe","amount":5000.0,"currency":"RUB", + "date":"…","dueDate":null,"note":"","repaid":1000.0,"isClosed":false} +``` + +`direction`: `iOwe` | `owedToMe`. Остаток = `amount − repaid`. Долги **не** +входят в баланс и аналитику: деньги, которые реально ушли или пришли, +пользователь записывает обычной операцией. + +### rate + +```json +{"from":"USD","to":"RUB","rate":"92.35","date":"2026-10-31T00:00:00Z"} +``` + +`rate` — сколько единиц `to` стоит одна единица `from`, до 6 знаков после +точки. + +### setting + +`id` — ключ, payload `{"value": …}`. Сейчас ключ один: `baseCurrency` +(строка ISO 4217) — валюта, в которой считаются общие суммы. Если записи нет, +базовая валюта — локальная настройка устройства (поведение v1). + +### attachment + +```json +{"transactionId":"…","sha256":"<64 hex>","contentType":"image/jpeg", + "size":182345,"name":"чек.jpg","createdAt":"…"} +``` + +Метаданные синхронизируются обычным `/v1/sync`; сами байты — через §8. + +## 2. Валюты и курсы + +- Валюта транзакции: `currency`, иначе валюта её счёта, иначе базовая. +- Перевод суммы `x` из `C` в базовую `B`: при `C == B` без изменений; иначе + берётся последняя (по `date`, затем по `updatedAt`) запись `rate` с + `from=C,to=B` или обратная `from=B,to=C` (тогда делим). Если подходящих + записей несколько с одной датой, прямая пара важнее обратной. +- Результат округляется до сотых половиной вверх (от нуля). Считать через + целые сотые и десятичную строку курса, не через `double`-накопление. +- Нет курса → сумма **не входит** в общие итоги, клиент показывает + предупреждение «нет курса C→B». Внутри счёта суммы складываются в валюте + счёта без конвертации. +- Курс для периода — последний на дату расчёта (не на дату операции). + +## 3. Баланс (расширяет финансовые правила v1) + +- Баланс счёта (в его валюте) = доходы − расходы с этим `accountId` + (включая корректировки) + `toAmount` входящих переводов − `amount` + исходящих. +- «Без счёта» = транзакции без `accountId` (в своих валютах). +- Общий баланс = сумма всех транзакций в базовой валюте + переводы + (входящая часть − исходящая, каждая в базовой валюте). При одной валюте + переводы дают ноль, и общий баланс совпадает с v1. +- Переводы не входят в аналитику и бюджеты. Корректировки — как в v1. +- Аналитика и бюджеты складывают суммы в базовой валюте. +- Архивный счёт скрыт из выбора, но его баланс входит в общий. +- Транзакции и переводы, ссылающиеся на удалённый счёт, остаются; клиент + показывает их как «Удалённый счёт». + +## 4. Регулярные операции + +- Повторения считаются в календаре **UTC** от даты `startDate` (её UTC-дата + и время суток): `daily` — каждые `interval` дней, `weekly` — каждые + `7·interval` дней, `monthly` — каждые `interval` месяцев в тот же день + (если дня нет — последний день месяца), `yearly` — то же с годами + (29 февраля → 28 февраля). Повторение после `endDate` не создаётся. +- `occurrence` = UTC-дата `YYYY-MM-DD`; дата транзакции = эта дата + время + суток `startDate`. +- ID транзакции **детерминирован**: UUID v5 от пространства имён + `6f1c3f5e-8a8d-4c3a-9d2b-4b1f0e7a2c11` и строки + `recurring::` (нижний регистр). Два устройства + создают одну и ту же запись — дублей нет. +- `auto`: клиент создаёт все повторения в `(generatedThrough, сегодня UTC]` + (с `startDate`, если `generatedThrough` пуст) и ставит + `generatedThrough` = последнее обработанное. Удалённое пользователем + повторение не воскрешается, т.к. `generatedThrough` уже позади. Не более + 366 повторений за раз. +- `remind`: клиент показывает повторения из того же окна как «ожидают»; + «Записать» создаёт транзакцию с тем же детерминированным ID, «Пропустить» + только двигает `generatedThrough`. +- `isPaused` — ничего не создаётся и не напоминается. +- Часы регулярные операции не создают (их создают телефон и десктоп). + +## 5. Цели, долги — см. §1. Цели и долги не меняют баланс. + +## 6. Теги + +Строки после `trim`, 1–32 символа, без учёта регистра уникальны в операции, +не больше 10 на операцию. Порядок сохраняется. Аналитика и бюджеты тегами не +фильтруются; список операций — фильтруется. + +## 7. Общие бюджеты (ledgers) + +Пространство данных (ledger) — отдельный набор записей со своим курсором. +У каждого пользователя есть **личное** пространство (как в v1). Общее +пространство создаёт владелец и приглашает участников по коду. + +| Метод | Путь | Тело / ответ | +| --- | --- | --- | +| GET | `/v1/ledgers` | `{"ledgers":[{"id","name","role","members":[{"userId","email","role"}]}]}` — только общие | +| POST | `/v1/ledgers` | `{"name"}` → ledger, `role:"owner"` | +| PATCH | `/v1/ledgers/{id}` | `{"name"}` (владелец) | +| DELETE | `/v1/ledgers/{id}` | владелец: удалить пространство со всеми записями и вложениями | +| POST | `/v1/ledgers/{id}/invites` | → `{"code","expiresAt"}`; код из 10 символов, 7 дней, одноразовый (владелец) | +| POST | `/v1/ledgers/join` | `{"code"}` → ledger, `role:"member"` | +| DELETE | `/v1/ledgers/{id}/members/{userId}` | владелец удаляет участника; участник — себя (`userId` = свой) | + +- Пространство выбирается заголовком `X-Balance-Ledger: ` на + `/v1/sync` и `/v1/attachments/*`. Без заголовка — личное пространство. + Не участник → `403 forbidden`, несуществующее → `404 not_found`. +- Владелец не может выйти, пока есть участники (только удалить пространство). +- Все участники равны в правах на записи (last-write-wins как в v1). +- Клиент хранит данные пространства отдельно: scope = сервер + пользователь + + пространство. Переключение пространства не переносит записи. Курсор и + очередь изменений — свои у каждого пространства. +- Ошибки: `invalid_code` (400), `forbidden` (403), `not_found` (404), + `owner_cannot_leave` (409). Коды приглашений хранятся на сервере только + как хеш. + +## 8. Вложения + +| Метод | Путь | Смысл | +| --- | --- | --- | +| PUT | `/v1/attachments/{sha256}` | тело — байты файла, `Content-Type` из списка; сервер проверяет SHA-256 и размер; повтор идемпотентен (`200`, первый раз `201`) | +| GET | `/v1/attachments/{sha256}` | байты и `Content-Type`; `404`, если в пространстве нет такого файла | + +- Порядок клиента: сначала PUT байтов, потом запись `attachment` через sync. +- Файлы хранятся в пространстве; один и тот же файл в двух пространствах + хранится дважды. +- Когда в пространстве не осталось живых записей `attachment` с этим + `sha256`, сервер может удалить файл. +- Ошибки: `payload_too_large` (413), `unsupported_media_type` (415), + `hash_mismatch` (400). +- Клиент скачивает файл лениво при просмотре и кеширует локально. + +## 9. Совместимость с клиентами v1 + +- Старый клиент не видит счета, переводы, цели, долги, курсы, вложения и + общие пространства; транзакции в других валютах складывает как есть. +- При редактировании транзакции старый клиент теряет `accountId`, + `currency`, `tags`, `recurringId`, `occurrence`. Поэтому все клиенты + экосистемы обновляются вместе с сервером. +- Нативные iOS/macOS и Android-клиенты (не входят в эти репозитории) остаются + на v1 до своего обновления. diff --git a/docs/DATA_MODEL.md b/docs/DATA_MODEL.md index 8a0e604..c1bb8e6 100644 --- a/docs/DATA_MODEL.md +++ b/docs/DATA_MODEL.md @@ -11,14 +11,16 @@ ## Общие сущности -Сервер должен различать как минимум следующие логические типы синхронизируемых данных, совместимые с клиентами Balance: +Сервер принимает сущности контракта v2 (`docs/CONTRACT_V2.md` §1): ```text -transaction -category -budget +transaction category budget account transfer recurring +goal debt rate setting attachment ``` +Payload хранится как непрозрачный JSON; сервер читает из него только `sha256` +записей `attachment` для сборки мусора файлов. Ниже — сущности v1. + ### Transaction Логически содержит: @@ -97,6 +99,30 @@ payload 7. Timestamps должны храниться в UTC. 8. Enum/string values должны быть совместимы с iOS и Android. +## Таблицы контракта v2 + +Таблицы v1 (`users`, `refresh_tokens`, `records`, `changes`) не меняются. Миграция +только добавляет таблицы через `CREATE TABLE IF NOT EXISTS`, поэтому существующая +база (в том числе Go v1) открывается без переписывания строк и `sequence`. + +| Таблица | Ключ | Назначение | +| --- | --- | --- | +| `ledgers` | `id` (UUID) | имя, `owner_id`, `created_at` | +| `ledger_members` | `(ledger_id, user_id)`, индекс `user_id` | `role` = `owner`/`member`, `joined_at` | +| `ledger_invites` | `code_hash` (SHA-256 кода), индекс `ledger_id` | `expires_at`, `used_at`, `used_by`; код в открытом виде не хранится | +| `ledger_records` | `(ledger_id, entity, record_id)` | текущее состояние записей пространства | +| `ledger_changes` | `sequence` AUTOINCREMENT, индекс `(ledger_id, sequence)` | поток изменений пространства | +| `attachment_blobs` | `(space, sha256)` | метаданные файлов: `content_type`, `size`, `uploaded_at`, `referenced` | + +`space` в `attachment_blobs` — `user:` или `ledger:`. Все таблицы +ledger ссылаются на `ledgers(id) ON DELETE CASCADE`; строки `attachment_blobs` +пространства удаляются в той же транзакции, что и ledger. + +Файлы лежат в `BALANCE_ATTACHMENTS_DIR` (по умолчанию `attachments` рядом с +файлом БД): `users///` и +`ledgers///`. Каталоги создаются с правами `0750`. +Один и тот же файл в двух пространствах хранится дважды. + ## Миграции Schema changes должны быть backward-compatible с активными версиями клиентов либо сопровождаться versioned API/миграцией. diff --git a/docs/DEPLOYMENT.md b/docs/DEPLOYMENT.md index 6e40d5a..43c73a3 100644 --- a/docs/DEPLOYMENT.md +++ b/docs/DEPLOYMENT.md @@ -35,6 +35,7 @@ for remote access; pass `X-Forwarded-Proto: https` to enable Secure session cook | `BALANCE_REFRESH_TTL` | `720h` | Go-style duration for refresh tokens | | `BALANCE_ALLOW_REGISTRATION` | `true` | Enable creation of accounts | | `BALANCE_CORS_ORIGINS` | empty | Comma-separated allowed browser origins | +| `BALANCE_ATTACHMENTS_DIR` | `attachments` beside the database | Attachment files; Docker uses `/data/attachments` inside the same volume. Back it up with the database | Inject secrets through the deployment environment or `.env`; never put them in images or source control. Compose preserves the existing named `balance-data` diff --git a/docs/SECURITY.md b/docs/SECURITY.md index d2cc3d2..15b4be3 100644 --- a/docs/SECURITY.md +++ b/docs/SECURITY.md @@ -51,6 +51,25 @@ HTTP следует оставлять только для controlled local deve - JSON size; - cursor/sequence bounds. +## Общие пространства и вложения (контракт v2) + +- Доступ к ledger проверяется на каждый запрос по `ledger_members`; `X-Balance-Ledger` + не участника → `403`, неизвестный → `404`. Управлять пространством и выдавать + приглашения может только владелец. +- Коды приглашений — 10 случайных символов из 32 (50 бит, `OsRng`), одноразовые, + 7 дней; в БД хранится только SHA-256. `/v1/ledgers/join` ограничен общим + лимитом 30 запросов/мин на IP. +- Вложения проверяются до записи на диск: формат хеша, тип из белого списка, + размер ≤ 10 MiB и SHA-256 тела. Файл пишется во временный файл и + переименовывается. Тело больше 2 MiB читается только после проверки + access token. Путь к файлу строится из проверенного hex-хеша и id + пространства (небезопасный id заменяется его хешем), поэтому значения клиента + не становятся компонентами пути. +- Файлы отдаются с `Content-Disposition: attachment`, `X-Content-Type-Options: + nosniff`, `Cache-Control: no-store` и не отображаются в origin веб-клиента. +- Каталог вложений содержит финансовые документы: права `0750`, резервные + копии — вместе с БД. + ## Logging Безопасно логировать: diff --git a/docs/SYNC.md b/docs/SYNC.md index 1205d90..40d4003 100644 --- a/docs/SYNC.md +++ b/docs/SYNC.md @@ -131,6 +131,35 @@ Server не должен предполагать, что все изменен Нельзя молча трактовать неизвестный/просроченный cursor как актуальный. +## Пространства (контракт v2) + +`POST /v1/sync` работает в одном пространстве данных: + +- без `X-Balance-Ledger` — личное пространство пользователя, таблицы + `records`/`changes`, поведение v1 без изменений; +- с `X-Balance-Ledger: ` — общее пространство, таблицы + `ledger_records`/`ledger_changes`. Доступ проверяется на каждый запрос. + +У общего пространства свой поток изменений: `sequence` берётся из отдельного +AUTOINCREMENT `ledger_changes`, а pull фильтрует по `ledger_id`. Курсор личного +пространства и курсор каждого ledger независимы, клиент хранит их раздельно. +Все участники пишут в одно пространство с тем же last-write-wins по +`updatedAt`; идемпотентность, пагинация (500), лимиты и транзакционность те же. + +Удаление пространства физически удаляет его записи и историю изменений. +Удалённый участник теряет доступ (`403`), записи остаются в пространстве. + +## Вложения и сборка мусора + +Байты вложений загружаются отдельно (`PUT /v1/attachments/{sha256}`) до записи +`attachment` в sync. Файл пространства считается мусором, когда ни одна живая +(`deleted=0`) запись `attachment` этого пространства не ссылается на его +`sha256` (сравнение без учёта регистра). Сборка выполняется в той же транзакции, +что и sync-запрос, содержащий изменения `attachment`, и при каждом PUT. +Файл, ни разу не получивший ссылки, хранится не меньше 24 часов с последней +загрузки, чтобы не удалить байты, чья запись ещё не синхронизирована; повторная +загрузка перезапускает этот срок. + ## Security Sync endpoint всегда ограничен текущим authenticated user. diff --git a/src/attachments.rs b/src/attachments.rs new file mode 100644 index 0000000..199a3c7 --- /dev/null +++ b/src/attachments.rs @@ -0,0 +1,326 @@ +//! Attachment blobs (contract v2, section 8). +//! +//! Bytes live on disk under the attachments directory, one subdirectory per +//! data space; their metadata lives in `attachment_blobs`. A blob is garbage +//! when no live `attachment` record of its space references its SHA-256. +//! Blobs that were uploaded but not referenced yet are kept for a grace period, +//! because clients upload the bytes before syncing the record. + +use crate::sync::Space; +use chrono::{DateTime, SecondsFormat, TimeDelta, Utc}; +use rusqlite::{Connection, OptionalExtension, params}; +use sha2::{Digest, Sha256}; +use std::{ + fs, + io::{self, Write}, + path::{Path, PathBuf}, +}; + +pub const MAX_BYTES: usize = 10 * 1024 * 1024; +pub const TYPES: [&str; 4] = ["image/jpeg", "image/png", "image/heic", "application/pdf"]; +/// How long an uploaded, never referenced blob survives garbage collection. +pub const GRACE: TimeDelta = TimeDelta::hours(24); + +pub struct Store { + dir: PathBuf, +} + +impl Store { + pub fn new(dir: impl Into) -> Self { + Self { dir: dir.into() } + } + + fn space_dir(&self, space: &Space) -> PathBuf { + let (kind, id) = match space { + Space::User(id) => ("users", id.as_str()), + Space::Ledger(id) => ("ledgers", id.as_str()), + }; + // Server-generated ids are hex/UUIDs; anything else is hashed so a + // database value can never become a path component such as "..". + let safe = !id.is_empty() + && id.len() <= 64 + && id + .bytes() + .all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_'); + let name = if safe { + id.to_owned() + } else { + crate::auth::token_hash(id) + }; + self.dir.join(kind).join(name) + } + + fn path(&self, space: &Space, sha: &str) -> PathBuf { + self.space_dir(space).join(&sha[..2]).join(sha) + } + + /// Write bytes to a temporary file of the space, ready to be renamed. + fn stage(&self, space: &Space, bytes: &[u8]) -> io::Result { + let dir = self.space_dir(space); + create_dir(&dir)?; + let name = crate::auth::identifier().map_err(io::Error::other)?; + let path = dir.join(format!(".upload-{name}")); + let mut file = fs::File::create(&path)?; + let written = file.write_all(bytes).and_then(|_| file.sync_all()); + if let Err(e) = written { + let _ = fs::remove_file(&path); + return Err(e); + } + Ok(path) + } + + fn install(&self, staged: &Path, space: &Space, sha: &str) -> io::Result<()> { + let path = self.path(space, sha); + if let Some(parent) = path.parent() { + create_dir(parent)?; + } + fs::rename(staged, path) + } + + pub fn read(&self, space: &Space, sha: &str) -> io::Result> { + fs::read(self.path(space, sha)) + } + + /// Remove blob files whose rows were already deleted. + pub fn remove(&self, space: &Space, shas: &[String]) { + for sha in shas { + if valid_sha(sha) { + match fs::remove_file(self.path(space, sha)) { + Ok(()) => {} + Err(e) if e.kind() == io::ErrorKind::NotFound => {} + Err(e) => eprintln!("attachment cleanup failed: {}", e.kind()), + } + } + } + } + + /// Remove every blob of a deleted space. + pub fn remove_space(&self, space: &Space) { + match fs::remove_dir_all(self.space_dir(space)) { + Ok(()) => {} + Err(e) if e.kind() == io::ErrorKind::NotFound => {} + Err(e) => eprintln!("attachment directory cleanup failed: {}", e.kind()), + } + } +} + +fn create_dir(path: &Path) -> io::Result<()> { + let mut builder = fs::DirBuilder::new(); + builder.recursive(true); + #[cfg(unix)] + { + use std::os::unix::fs::DirBuilderExt; + builder.mode(0o750); + } + builder.create(path) +} + +/// SHA-256 in the canonical form of the contract: 64 lowercase hex digits. +pub fn valid_sha(sha: &str) -> bool { + sha.len() == 64 && sha.bytes().all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f')) +} + +/// Accepted media type of a `Content-Type` header, ignoring parameters. +pub fn media_type(header: &str) -> Option<&'static str> { + let essence = header.split(';').next()?.trim().to_ascii_lowercase(); + TYPES.into_iter().find(|t| *t == essence) +} + +fn stamp(t: DateTime) -> String { + // Fixed width, so stored stamps compare correctly as text. + t.to_rfc3339_opts(SecondsFormat::Nanos, true) +} + +fn live_shas(space: &Space) -> String { + let (records, _, key) = space.tables(); + format!( + "SELECT sha FROM (SELECT lower(CASE WHEN json_valid(payload) THEN json_extract(payload,'$.sha256') END) AS sha FROM {records} WHERE {key}=?2 AND entity='attachment' AND deleted=0) WHERE sha IS NOT NULL" + ) +} + +/// Delete rows of blobs no live attachment record references and return +/// their hashes. Blobs never referenced survive for [`GRACE`] after upload. +pub fn collect_garbage( + db: &Connection, + space: &Space, + now: DateTime, +) -> rusqlite::Result> { + let live = live_shas(space); + let key = space.key(); + db.execute( + &format!( + "UPDATE attachment_blobs SET referenced=1 WHERE space=?1 AND referenced=0 AND sha256 IN ({live})" + ), + params![key, space.id()], + )?; + let condition = + format!("space=?1 AND (referenced=1 OR uploaded_at>>()?; + db.execute( + &format!("DELETE FROM attachment_blobs WHERE {condition}"), + params![key, space.id(), stamp(now - GRACE)], + )?; + Ok(garbage) +} + +#[derive(Debug, PartialEq, Eq)] +pub enum PutError { + HashMismatch, + UnsupportedMediaType, + PayloadTooLarge, + Internal, +} + +/// Store bytes in `space`. Returns `true` when the blob was created and +/// `false` for an idempotent repeat. The database lock is taken only to +/// install the already written file. +pub fn put( + db: &std::sync::Mutex, + store: &Store, + space: &Space, + sha: &str, + content_type: &str, + bytes: &[u8], +) -> Result { + if !valid_sha(sha) { + return Err(PutError::HashMismatch); + } + let Some(media) = media_type(content_type) else { + return Err(PutError::UnsupportedMediaType); + }; + if bytes.len() > MAX_BYTES { + return Err(PutError::PayloadTooLarge); + } + let digest: String = Sha256::digest(bytes) + .iter() + .map(|b| format!("{b:02x}")) + .collect(); + if digest != sha { + return Err(PutError::HashMismatch); + } + let staged = store.stage(space, bytes).map_err(|_| PutError::Internal)?; + let result = install(db, store, space, sha, media, bytes.len(), &staged); + let _ = fs::remove_file(&staged); + result +} + +fn install( + db: &std::sync::Mutex, + store: &Store, + space: &Space, + sha: &str, + media: &str, + size: usize, + staged: &Path, +) -> Result { + let internal = |_| PutError::Internal; + let mut db = db.lock().unwrap_or_else(|e| e.into_inner()); + let tx = db.transaction().map_err(internal)?; + let now = Utc::now(); + let key = space.key(); + let exists = tx + .query_row( + "SELECT 1 FROM attachment_blobs WHERE space=? AND sha256=?", + params![key, sha], + |_| Ok(()), + ) + .optional() + .map_err(internal)? + .is_some(); + let path = store.path(space, sha); + let file_missing = !path.is_file(); + if exists { + // A repeated upload restarts the grace period: another device may + // be about to sync a new record that references these bytes. + tx.execute( + "UPDATE attachment_blobs SET uploaded_at=?, referenced=0 WHERE space=? AND sha256=?", + params![stamp(now), key, sha], + ) + .map_err(internal)?; + } else { + tx.execute( + "INSERT INTO attachment_blobs(space,sha256,content_type,size,uploaded_at,referenced) VALUES(?,?,?,?,?,0)", + params![key, sha, media, size as i64, stamp(now)], + ) + .map_err(internal)?; + } + let garbage = collect_garbage(&tx, space, now).map_err(internal)?; + if file_missing { + store + .install(staged, space, sha) + .map_err(|_| PutError::Internal)?; + } + if tx.commit().is_err() { + if file_missing { + let _ = fs::remove_file(&path); + } + return Err(PutError::Internal); + } + store.remove(space, &garbage); + Ok(!exists) +} + +/// Bytes and media type of a blob stored in `space`. +pub fn get( + db: &std::sync::Mutex, + store: &Store, + space: &Space, + sha: &str, +) -> Result, String)>, ()> { + if !valid_sha(sha) { + return Ok(None); + } + let db = db.lock().unwrap_or_else(|e| e.into_inner()); + let media: Option = db + .query_row( + "SELECT content_type FROM attachment_blobs WHERE space=? AND sha256=?", + params![space.key(), sha], + |r| r.get(0), + ) + .optional() + .map_err(|_| ())?; + let Some(media) = media else { + return Ok(None); + }; + // Reading under the lock keeps garbage collection from removing the + // file between the metadata check and the read. + match store.read(space, sha) { + Ok(bytes) => Ok(Some((bytes, media))), + Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(None), + Err(_) => Err(()), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn hashes_and_media_types_are_strict() { + assert!(valid_sha(&"a".repeat(64))); + assert!(!valid_sha(&"A".repeat(64))); + assert!(!valid_sha(&"a".repeat(63))); + assert!(!valid_sha(&format!("{}g", "a".repeat(63)))); + assert_eq!(media_type("IMAGE/JPEG; q=1"), Some("image/jpeg")); + assert_eq!(media_type("application/pdf"), Some("application/pdf")); + assert_eq!(media_type("image/gif"), None); + assert_eq!(media_type(""), None); + } + + #[test] + fn unsafe_space_ids_never_become_path_components() { + let store = Store::new("/base"); + let dir = store.space_dir(&Space::Ledger("../../etc".into())); + assert!(dir.starts_with("/base/ledgers")); + assert!(!dir.to_string_lossy().contains("..")); + assert_eq!( + store.space_dir(&Space::User("abc".into())), + Path::new("/base/users/abc") + ); + } +} diff --git a/src/config.rs b/src/config.rs index 7d99fc7..c8055b4 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1,5 +1,8 @@ use chrono::TimeDelta; -use std::env; +use std::{ + env, + path::{Path, PathBuf}, +}; #[derive(Clone, Debug)] pub struct Config { @@ -10,6 +13,7 @@ pub struct Config { pub refresh_ttl: TimeDelta, pub allow_registration: bool, pub cors_origins: Vec, + pub attachments_dir: PathBuf, } impl Config { @@ -25,9 +29,14 @@ impl Config { if jwt_secret.len() < 32 { return Err("BALANCE_JWT_SECRET must contain at least 32 characters".into()); } + let database_path = fallback("BALANCE_DB_PATH", "./data/balance.db"); + let attachments_dir = match value("BALANCE_ATTACHMENTS_DIR") { + dir if dir.is_empty() => default_attachments_dir(&database_path), + dir => PathBuf::from(dir), + }; Ok(Self { address: fallback("BALANCE_ADDR", ":8080"), - database_path: fallback("BALANCE_DB_PATH", "./data/balance.db"), + database_path, jwt_secret, access_ttl, refresh_ttl, @@ -38,10 +47,19 @@ impl Config { .filter(|part| !part.is_empty()) .map(str::to_owned) .collect(), + attachments_dir, }) } } +/// `attachments` next to the database file. +pub fn default_attachments_dir(database_path: &str) -> PathBuf { + Path::new(database_path) + .parent() + .unwrap_or(Path::new("")) + .join("attachments") +} + fn value(key: &str) -> String { env::var(key).unwrap_or_default().trim().to_owned() } @@ -128,6 +146,17 @@ pub fn parse_go_duration(input: &str) -> Option { mod tests { use super::*; #[test] + fn attachments_default_next_to_database() { + assert_eq!( + default_attachments_dir("/data/balance.db"), + Path::new("/data/attachments") + ); + assert_eq!( + default_attachments_dir("balance.db"), + Path::new("attachments") + ); + } + #[test] fn duration_matches_go_syntax_and_range() { assert_eq!(parse_go_duration("1h30m"), Some(TimeDelta::minutes(90))); assert_eq!( diff --git a/src/database.rs b/src/database.rs index 07b9369..f22d220 100644 --- a/src/database.rs +++ b/src/database.rs @@ -1,7 +1,8 @@ use rusqlite::Connection; use std::{error::Error, path::Path, time::Duration}; -/// Open the existing Go server's SQLite database without changing its schema. +/// Open the SQLite database. Tables of the Go server's v1 schema are never +/// altered; contract v2 only adds new tables (ledgers, attachments). pub fn open(path: &str) -> Result> { if let Some(parent) = Path::new(path) .parent() @@ -63,7 +64,61 @@ CREATE TABLE IF NOT EXISTS changes ( deleted INTEGER NOT NULL, version INTEGER NOT NULL ); -CREATE INDEX IF NOT EXISTS changes_user_sequence_idx ON changes(user_id, sequence);", +CREATE INDEX IF NOT EXISTS changes_user_sequence_idx ON changes(user_id, sequence); +CREATE TABLE IF NOT EXISTS ledgers ( + id TEXT PRIMARY KEY, + name TEXT NOT NULL, + owner_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE, + created_at TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS ledger_members ( + ledger_id TEXT NOT NULL REFERENCES ledgers(id) ON DELETE CASCADE, + user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE, + role TEXT NOT NULL CHECK (role IN ('owner','member')), + joined_at TEXT NOT NULL, + PRIMARY KEY (ledger_id, user_id) +); +CREATE INDEX IF NOT EXISTS ledger_members_user_idx ON ledger_members(user_id); +CREATE TABLE IF NOT EXISTS ledger_invites ( + code_hash TEXT PRIMARY KEY, + ledger_id TEXT NOT NULL REFERENCES ledgers(id) ON DELETE CASCADE, + created_by TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE, + expires_at TEXT NOT NULL, + created_at TEXT NOT NULL, + used_at TEXT, + used_by TEXT REFERENCES users(id) ON DELETE SET NULL +); +CREATE INDEX IF NOT EXISTS ledger_invites_ledger_idx ON ledger_invites(ledger_id); +CREATE TABLE IF NOT EXISTS ledger_records ( + ledger_id TEXT NOT NULL REFERENCES ledgers(id) ON DELETE CASCADE, + entity TEXT NOT NULL, + record_id TEXT NOT NULL, + payload TEXT, + updated_at TEXT NOT NULL, + deleted INTEGER NOT NULL DEFAULT 0, + version INTEGER NOT NULL, + PRIMARY KEY (ledger_id, entity, record_id) +); +CREATE TABLE IF NOT EXISTS ledger_changes ( + sequence INTEGER PRIMARY KEY AUTOINCREMENT, + ledger_id TEXT NOT NULL REFERENCES ledgers(id) ON DELETE CASCADE, + entity TEXT NOT NULL, + record_id TEXT NOT NULL, + payload TEXT, + updated_at TEXT NOT NULL, + deleted INTEGER NOT NULL, + version INTEGER NOT NULL +); +CREATE INDEX IF NOT EXISTS ledger_changes_ledger_sequence_idx ON ledger_changes(ledger_id, sequence); +CREATE TABLE IF NOT EXISTS attachment_blobs ( + space TEXT NOT NULL, + sha256 TEXT NOT NULL, + content_type TEXT NOT NULL, + size INTEGER NOT NULL, + uploaded_at TEXT NOT NULL, + referenced INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (space, sha256) +);", ) } @@ -155,6 +210,26 @@ mod tests { .unwrap(), "{\"amount\":12.5}" ); + // Contract v2 tables are added next to the untouched v1 tables. + for table in [ + "ledgers", + "ledger_members", + "ledger_invites", + "ledger_records", + "ledger_changes", + "attachment_blobs", + ] { + assert_eq!( + db.query_row( + "SELECT count(*) FROM sqlite_master WHERE type='table' AND name=?", + [table], + |r| r.get::<_, i64>(0) + ) + .unwrap(), + 1, + "{table}" + ); + } db.execute("INSERT INTO changes(user_id,entity,record_id,payload,updated_at,deleted,version) VALUES ('go-user','transaction','next',NULL,'now',1,42)", []).unwrap(); assert_eq!(db.last_insert_rowid(), 42); assert_eq!( diff --git a/src/ledgers.rs b/src/ledgers.rs new file mode 100644 index 0000000..1e0876d --- /dev/null +++ b/src/ledgers.rs @@ -0,0 +1,365 @@ +//! Shared ledgers (contract v2, section 7). +//! +//! A ledger is a data space with its own records, change stream and +//! attachments. Membership is checked for every request that selects it. + +use crate::sync::Space; +use chrono::{SecondsFormat, TimeDelta, Utc}; +use rusqlite::{Connection, OptionalExtension, params}; +use serde_json::{Value, json}; + +pub type LedgerError = (u16, &'static str, &'static str); + +pub const INVITE_TTL: TimeDelta = TimeDelta::days(7); +pub const CODE_LENGTH: usize = 10; +/// 32 symbols without look-alikes (no I, O, 0, 1): five random bits each. +const CODE_ALPHABET: &[u8; 32] = b"ABCDEFGHJKLMNPQRSTUVWXYZ23456789"; +const MAX_NAME: usize = 100; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Role { + Owner, + Member, +} + +impl Role { + fn as_str(self) -> &'static str { + match self { + Role::Owner => "owner", + Role::Member => "member", + } + } +} + +fn internal(_: E) -> LedgerError { + ( + 500, + "internal_error", + "Could not process the ledger request", + ) +} +fn not_found() -> LedgerError { + (404, "not_found", "Ledger not found") +} +fn forbidden() -> LedgerError { + (403, "forbidden", "Not allowed for this ledger") +} +fn invalid_code() -> LedgerError { + (400, "invalid_code", "Invite code is invalid or expired") +} + +fn now() -> String { + Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true) +} + +/// Role of `user` in `ledger`: 404 for an unknown ledger, 403 for a non-member. +pub fn role(db: &Connection, ledger: &str, user: &str) -> Result { + let exists = db + .query_row("SELECT 1 FROM ledgers WHERE id=?", [ledger], |_| Ok(())) + .optional() + .map_err(internal)? + .is_some(); + if !exists { + return Err(not_found()); + } + let role: Option = db + .query_row( + "SELECT role FROM ledger_members WHERE ledger_id=? AND user_id=?", + params![ledger, user], + |r| r.get(0), + ) + .optional() + .map_err(internal)?; + match role.as_deref() { + Some("owner") => Ok(Role::Owner), + Some(_) => Ok(Role::Member), + None => Err(forbidden()), + } +} + +/// Data space selected by the `X-Balance-Ledger` header value. +pub fn space(db: &Connection, header: &str, user: &str) -> Result { + let ledger = header.trim(); + if ledger.is_empty() { + return Ok(Space::User(user.to_owned())); + } + role(db, ledger, user)?; + Ok(Space::Ledger(ledger.to_owned())) +} + +fn describe(db: &Connection, ledger: &str, role: Role) -> Result { + let name: String = db + .query_row("SELECT name FROM ledgers WHERE id=?", [ledger], |r| { + r.get(0) + }) + .map_err(internal)?; + let mut stmt = db + .prepare( + "SELECT m.user_id,u.email,m.role FROM ledger_members m JOIN users u ON u.id=m.user_id + WHERE m.ledger_id=? ORDER BY m.role='owner' DESC, m.joined_at, m.user_id", + ) + .map_err(internal)?; + let members = stmt + .query_map([ledger], |r| { + Ok(json!({"userId": r.get::<_, String>(0)?, "email": r.get::<_, String>(1)?, "role": r.get::<_, String>(2)?})) + }) + .map_err(internal)? + .collect::>>() + .map_err(internal)?; + Ok(json!({"id": ledger, "name": name, "role": role.as_str(), "members": members})) +} + +/// Shared ledgers of `user`; the personal space is not listed. +pub fn list(db: &Connection, user: &str) -> Result { + let mut stmt = db + .prepare( + "SELECT l.id,m.role FROM ledger_members m JOIN ledgers l ON l.id=m.ledger_id + WHERE m.user_id=? ORDER BY l.created_at, l.id", + ) + .map_err(internal)?; + let rows = stmt + .query_map([user], |r| { + Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)? == "owner")) + }) + .map_err(internal)? + .collect::>>() + .map_err(internal)?; + let ledgers = rows + .into_iter() + .map(|(id, owner)| describe(db, &id, if owner { Role::Owner } else { Role::Member })) + .collect::, _>>()?; + Ok(json!({ "ledgers": ledgers })) +} + +/// Ledger name from a `{"name": …}` body: trimmed, 1–100 characters. +pub fn parse_name(body: &[u8]) -> Result { + let value: Value = serde_json::from_slice(body) + .map_err(|_| (400, "invalid_json", "Request body is not valid JSON"))?; + let name = value + .get("name") + .and_then(Value::as_str) + .map(str::trim) + .unwrap_or(""); + if name.is_empty() || name.chars().count() > MAX_NAME || name.chars().any(char::is_control) { + return Err(( + 400, + "invalid_name", + "Ledger name must contain 1 to 100 characters", + )); + } + Ok(name.to_owned()) +} + +pub fn create(db: &mut Connection, user: &str, name: &str) -> Result { + let id = uuid_v4().map_err(internal)?; + let tx = db.transaction().map_err(internal)?; + let created = now(); + tx.execute( + "INSERT INTO ledgers(id,name,owner_id,created_at) VALUES(?,?,?,?)", + params![id, name, user, created], + ) + .map_err(internal)?; + tx.execute( + "INSERT INTO ledger_members(ledger_id,user_id,role,joined_at) VALUES(?,?,'owner',?)", + params![id, user, created], + ) + .map_err(internal)?; + let ledger = describe(&tx, &id, Role::Owner)?; + tx.commit().map_err(internal)?; + Ok(ledger) +} + +pub fn rename(db: &Connection, ledger: &str, user: &str, name: &str) -> Result { + if role(db, ledger, user)? != Role::Owner { + return Err(forbidden()); + } + db.execute( + "UPDATE ledgers SET name=? WHERE id=?", + params![name, ledger], + ) + .map_err(internal)?; + describe(db, ledger, Role::Owner) +} + +/// Delete a ledger with its records, changes, invites and blob rows. The +/// caller removes the blob directory afterwards. +pub fn delete(db: &mut Connection, ledger: &str, user: &str) -> Result<(), LedgerError> { + if role(db, ledger, user)? != Role::Owner { + return Err(forbidden()); + } + let tx = db.transaction().map_err(internal)?; + tx.execute( + "DELETE FROM attachment_blobs WHERE space=?", + [Space::Ledger(ledger.to_owned()).key()], + ) + .map_err(internal)?; + // Cascades to members, invites, records and changes. + tx.execute("DELETE FROM ledgers WHERE id=?", [ledger]) + .map_err(internal)?; + tx.commit().map_err(internal) +} + +pub fn invite(db: &Connection, ledger: &str, user: &str) -> Result { + if role(db, ledger, user)? != Role::Owner { + return Err(forbidden()); + } + let code = random_code().map_err(internal)?; + let created = Utc::now(); + let expires = created + INVITE_TTL; + db.execute( + "INSERT INTO ledger_invites(code_hash,ledger_id,created_by,expires_at,created_at) VALUES(?,?,?,?,?)", + params![ + crate::auth::token_hash(&code), + ledger, + user, + expires.to_rfc3339_opts(SecondsFormat::Nanos, true), + created.to_rfc3339_opts(SecondsFormat::Nanos, true) + ], + ) + .map_err(internal)?; + Ok(json!({ + "code": code, + "expiresAt": expires.to_rfc3339_opts(SecondsFormat::Secs, true) + })) +} + +/// Canonical invite code: case-insensitive, spaces and dashes ignored. +pub fn normalize_code(code: &str) -> Option { + let code: String = code + .chars() + .filter(|c| !c.is_whitespace() && *c != '-') + .map(|c| c.to_ascii_uppercase()) + .collect(); + (code.len() == CODE_LENGTH && code.bytes().all(|b| CODE_ALPHABET.contains(&b))).then_some(code) +} + +pub fn join(db: &mut Connection, user: &str, body: &[u8]) -> Result { + let value: Value = serde_json::from_slice(body) + .map_err(|_| (400, "invalid_json", "Request body is not valid JSON"))?; + let code = value + .get("code") + .and_then(Value::as_str) + .and_then(normalize_code) + .ok_or_else(invalid_code)?; + let tx = db.transaction().map_err(internal)?; + let invite: Option<(String, String, Option)> = tx + .query_row( + "SELECT ledger_id,expires_at,used_at FROM ledger_invites WHERE code_hash=?", + [crate::auth::token_hash(&code)], + |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)), + ) + .optional() + .map_err(internal)?; + let Some((ledger, expires, used)) = invite else { + return Err(invalid_code()); + }; + if used.is_some() || expires <= now() { + return Err(invalid_code()); + } + // Joining a ledger one already belongs to does not consume the code. + if let Ok(existing) = role(&tx, &ledger, user) { + return describe(&tx, &ledger, existing); + } + let joined = now(); + tx.execute( + "UPDATE ledger_invites SET used_at=?, used_by=? WHERE code_hash=?", + params![joined, user, crate::auth::token_hash(&code)], + ) + .map_err(internal)?; + tx.execute( + "INSERT INTO ledger_members(ledger_id,user_id,role,joined_at) VALUES(?,?,'member',?)", + params![ledger, user, joined], + ) + .map_err(internal)?; + let result = describe(&tx, &ledger, Role::Member)?; + tx.commit().map_err(internal)?; + Ok(result) +} + +/// The owner removes a member; a member removes only itself. +pub fn remove_member( + db: &Connection, + ledger: &str, + user: &str, + target: &str, +) -> Result<(), LedgerError> { + let caller = role(db, ledger, user)?; + if caller == Role::Member && target != user { + return Err(forbidden()); + } + let target_role = match role(db, ledger, target) { + Ok(r) => r, + Err((403, _, _)) => return Err((404, "not_found", "Member not found")), + Err(e) => return Err(e), + }; + if target_role == Role::Owner { + // Leaving is impossible while the ledger has an owner; the owner + // deletes the ledger instead. + return Err(( + 409, + "owner_cannot_leave", + "The owner cannot leave the ledger; delete it instead", + )); + } + db.execute( + "DELETE FROM ledger_members WHERE ledger_id=? AND user_id=?", + params![ledger, target], + ) + .map_err(internal)?; + Ok(()) +} + +fn random_code() -> Result { + let mut bytes = [0u8; CODE_LENGTH]; + getrandom(&mut bytes)?; + Ok(bytes + .iter() + .map(|b| CODE_ALPHABET[usize::from(b & 31)] as char) + .collect()) +} + +fn uuid_v4() -> Result { + let mut b = [0u8; 16]; + getrandom(&mut b)?; + b[6] = (b[6] & 0x0f) | 0x40; + b[8] = (b[8] & 0x3f) | 0x80; + let hex: String = b.iter().map(|v| format!("{v:02x}")).collect(); + Ok(format!( + "{}-{}-{}-{}-{}", + &hex[..8], + &hex[8..12], + &hex[12..16], + &hex[16..20], + &hex[20..] + )) +} + +fn getrandom(bytes: &mut [u8]) -> Result<(), String> { + use rand::RngCore; + rand::rngs::OsRng + .try_fill_bytes(bytes) + .map_err(|_| "random source unavailable".to_owned()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn codes_are_ten_unambiguous_symbols_and_normalize() { + let code = random_code().unwrap(); + assert_eq!(code.len(), CODE_LENGTH); + assert_eq!(normalize_code(&code), Some(code.clone())); + let lower = format!( + " {}-{} ", + code[..5].to_lowercase(), + code[5..].to_lowercase() + ); + assert_eq!(normalize_code(&lower), Some(code)); + assert_eq!(normalize_code("ABCDEFGHI0"), None); + assert_eq!(normalize_code("ABCDEFGHJ"), None); + let id = uuid_v4().unwrap(); + assert_eq!(id.len(), 36); + assert_eq!(&id[14..15], "4"); + } +} diff --git a/src/main.rs b/src/main.rs index 666eb3f..dd1fab4 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,7 +1,9 @@ +mod attachments; mod auth; mod config; mod database; mod go_json; +mod ledgers; mod runtime; mod server; mod sync; diff --git a/src/server.rs b/src/server.rs index b6dc94c..4a431db 100644 --- a/src/server.rs +++ b/src/server.rs @@ -1,4 +1,4 @@ -use crate::{auth, config::Config, sync}; +use crate::{attachments, auth, config::Config, ledgers, sync}; use axum::{ body::{Body, HttpBody}, extract::{ConnectInfo, State as ExtractState}, @@ -24,12 +24,14 @@ pub struct State { cfg: Config, db: Mutex, tokens: auth::TokenManager, + attachments: attachments::Store, clients: Mutex>, } impl State { pub fn new(cfg: Config, db: Mutex) -> Self { Self { tokens: auth::TokenManager::new(&cfg.jwt_secret, cfg.access_ttl), + attachments: attachments::Store::new(cfg.attachments_dir.clone()), cfg, db, clients: Mutex::new(HashMap::new()), @@ -62,8 +64,17 @@ pub async fn handle( }; redirect(&location, &method) } else { - // Bound memory independently of Content-Length, including chunked requests. - let body = read_body(body).await; + // Bound memory independently of Content-Length, including chunked + // requests. Only an authorized attachment upload may exceed 2 MiB. + let limit = if method == Method::PUT + && percent_decode(&path).starts_with("/v1/attachments/") + && authorize(&state, &headers).is_ok() + { + attachments::MAX_BYTES + 1 + } else { + 2 << 20 + }; + let body = read_body(body, limit).await; let state = state.clone(); let headers = headers.clone(); let escaped_slash = path.to_ascii_lowercase().contains("%2f"); @@ -134,13 +145,25 @@ pub async fn handle( h.insert("Access-Control-Allow-Origin", value); } h.insert("Vary", HeaderValue::from_static("Origin")); + // v1 responses keep the recorded Go headers; contract v2 routes and + // real preflights (which v1 browsers also send) get the superset. + let extended = is_v2_route(&percent_decode(&path)) + || headers.contains_key("access-control-request-method"); h.insert( "Access-Control-Allow-Headers", - HeaderValue::from_static("Authorization, Content-Type"), + HeaderValue::from_static(if extended { + "Authorization, Content-Type, X-Balance-Ledger" + } else { + "Authorization, Content-Type" + }), ); h.insert( "Access-Control-Allow-Methods", - HeaderValue::from_static("GET, POST, OPTIONS"), + HeaderValue::from_static(if extended { + "GET, POST, PUT, PATCH, DELETE, OPTIONS" + } else { + "GET, POST, OPTIONS" + }), ); } if is_head { @@ -151,16 +174,16 @@ pub async fn handle( // Go's decoder consumes one value, so a valid first value may be followed by // ignored data beyond MaxBytesReader's cap. Keep the bounded prefix available. -async fn read_body(mut body: Body) -> Result, ()> { +async fn read_body(mut body: Body, limit: usize) -> Result, ()> { let mut bytes = Vec::new(); - while bytes.len() < 2 << 20 { + while bytes.len() < limit { let frame = std::future::poll_fn(|cx| std::pin::Pin::new(&mut body).poll_frame(cx)).await; let Some(frame) = frame else { break; }; let frame = frame.map_err(|_| ())?; if let Ok(data) = frame.into_data() { - let remaining = (2 << 20) - bytes.len(); + let remaining = limit - bytes.len(); bytes.extend_from_slice(&data[..data.len().min(remaining)]); } } @@ -175,6 +198,9 @@ fn dispatch( body: Result<&[u8], &()>, host: &str, ) -> Response { + if let Some(response) = dispatch_v2(state, method, path, headers, body, host) { + return response; + } let post_route = matches!( path, "/v1/auth/register" @@ -208,20 +234,9 @@ fn dispatch( if matches!( path, "/v1/auth/register" | "/v1/auth/login" | "/v1/auth/refresh" - ) { - let now = Instant::now(); - let mut clients = state.clients.lock().unwrap_or_else(|e| e.into_inner()); - let bucket = clients.entry(host.to_owned()).or_insert((now, 0)); - if now.duration_since(bucket.0) >= Duration::from_secs(60) { - *bucket = (now, 0); - } - bucket.1 = bucket.1.saturating_add(1); - if bucket.1 > 30 { - let mut r = error(429, "rate_limited", "Too many authentication requests"); - r.headers_mut() - .insert("Retry-After", HeaderValue::from_static("60")); - return r; - } + ) && let Some(r) = rate_limited(state, host) + { + return r; } let claims = if path == "/v1/sync" { match authorize(state, headers) { @@ -238,12 +253,17 @@ fn dispatch( return invalid_json(); }; if let Some(claims) = claims { - return match sync::synchronize_wire( - &mut state.db.lock().unwrap_or_else(|e| e.into_inner()), - &claims.subject, - body, - ) { - Ok(value) => raw_json_response(200, value), + let mut db = state.db.lock().unwrap_or_else(|e| e.into_inner()); + let space = match ledgers::space(&db, header(headers, LEDGER_HEADER), &claims.subject) { + Ok(space) => space, + Err((status, code, message)) => return error(status, code, message), + }; + return match sync::synchronize_space(&mut db, &space, body) { + Ok((value, garbage)) => { + // Still under the database lock, so no upload can race. + state.attachments.remove(&space, &garbage); + raw_json_response(200, value) + } Err((status, code, message)) => error(status, code, &message), }; } @@ -280,6 +300,248 @@ fn dispatch( } } +const LEDGER_HEADER: &str = "x-balance-ledger"; + +/// Shared limiter of endpoints that accept guessable secrets. +fn rate_limited(state: &State, host: &str) -> Option { + let now = Instant::now(); + let mut clients = state.clients.lock().unwrap_or_else(|e| e.into_inner()); + let bucket = clients.entry(host.to_owned()).or_insert((now, 0)); + if now.duration_since(bucket.0) >= Duration::from_secs(60) { + *bucket = (now, 0); + } + bucket.1 = bucket.1.saturating_add(1); + if bucket.1 > 30 { + let mut r = error(429, "rate_limited", "Too many authentication requests"); + r.headers_mut() + .insert("Retry-After", HeaderValue::from_static("60")); + return Some(r); + } + None +} + +enum Route<'a> { + Capabilities, + Ledgers, + Join, + Ledger(&'a str), + Invites(&'a str), + Member(&'a str, &'a str), + Attachment(&'a str), +} + +fn route(path: &str) -> Option> { + let rest = path.strip_prefix("/v1/")?; + let segments: Vec<&str> = rest.split('/').collect(); + if segments.iter().any(|s| s.is_empty()) { + return None; + } + Some(match segments.as_slice() { + ["capabilities"] => Route::Capabilities, + ["ledgers"] => Route::Ledgers, + ["ledgers", "join"] => Route::Join, + ["ledgers", id] => Route::Ledger(id), + ["ledgers", id, "invites"] => Route::Invites(id), + ["ledgers", id, "members", user] => Route::Member(id, user), + ["attachments", sha] => Route::Attachment(sha), + _ => return None, + }) +} + +fn is_v2_route(path: &str) -> bool { + route(path).is_some() +} + +fn method_not_allowed(allow: &'static str) -> Response { + let mut r = plain(405, "Method Not Allowed\n"); + r.headers_mut() + .insert("Allow", HeaderValue::from_static(allow)); + r +} + +fn ledger_error((status, code, message): ledgers::LedgerError) -> Response { + error(status, code, message) +} + +/// Contract v2 endpoints. `None` hands the request to the v1 dispatcher. +fn dispatch_v2( + state: &State, + method: &Method, + path: &str, + headers: &HeaderMap, + body: Result<&[u8], &()>, + host: &str, +) -> Option { + let route = route(path)?; + let get = method == Method::GET || method == Method::HEAD; + let allowed = match route { + Route::Capabilities => get, + Route::Ledgers => get || method == Method::POST, + Route::Join | Route::Invites(_) => method == Method::POST, + Route::Ledger(_) => method == Method::PATCH || method == Method::DELETE, + Route::Member(..) => method == Method::DELETE, + Route::Attachment(_) => get || method == Method::PUT, + }; + if !allowed { + return Some(method_not_allowed(match route { + Route::Capabilities => "GET, HEAD", + Route::Ledgers => "GET, HEAD, POST", + Route::Join | Route::Invites(_) => "POST", + Route::Ledger(_) => "PATCH, DELETE", + Route::Member(..) => "DELETE", + Route::Attachment(_) => "GET, HEAD, PUT", + })); + } + if let Route::Capabilities = route { + return Some(json_response(200, capabilities())); + } + if let Route::Join = route + && let Some(r) = rate_limited(state, host) + { + return Some(r); + } + let claims = match authorize(state, headers) { + Ok(c) => c, + Err(r) => return Some(*r), + }; + let user = claims.subject.as_str(); + let Ok(body) = body else { + return Some(invalid_json()); + }; + let lock = || state.db.lock().unwrap_or_else(|e| e.into_inner()); + Some(match route { + Route::Capabilities => unreachable!("answered above"), + Route::Ledgers if get => match ledgers::list(&lock(), user) { + Ok(v) => json_response(200, v), + Err(e) => ledger_error(e), + }, + Route::Ledgers => match ledgers::parse_name(body) + .and_then(|name| ledgers::create(&mut lock(), user, &name)) + { + Ok(v) => json_response(201, v), + Err(e) => ledger_error(e), + }, + Route::Join => match ledgers::join(&mut lock(), user, body) { + Ok(v) => json_response(200, v), + Err(e) => ledger_error(e), + }, + Route::Ledger(id) if method == Method::PATCH => { + let db = lock(); + match ledgers::role(&db, id, user) + .and_then(|_| ledgers::parse_name(body)) + .and_then(|name| ledgers::rename(&db, id, user, &name)) + { + Ok(v) => json_response(200, v), + Err(e) => ledger_error(e), + } + } + Route::Ledger(id) => { + let mut db = lock(); + match ledgers::delete(&mut db, id, user) { + Ok(()) => { + // Still under the lock: no upload can recreate the directory. + state + .attachments + .remove_space(&sync::Space::Ledger(id.to_owned())); + empty(204) + } + Err(e) => ledger_error(e), + } + } + Route::Invites(id) => match ledgers::invite(&lock(), id, user) { + Ok(v) => json_response(201, v), + Err(e) => ledger_error(e), + }, + Route::Member(id, target) => match ledgers::remove_member(&lock(), id, user, target) { + Ok(()) => empty(204), + Err(e) => ledger_error(e), + }, + Route::Attachment(sha) => { + let space = match ledgers::space(&lock(), header(headers, LEDGER_HEADER), user) { + Ok(space) => space, + Err(e) => return Some(ledger_error(e)), + }; + if get { + get_attachment(state, &space, sha) + } else { + put_attachment(state, &space, sha, header(headers, "content-type"), body) + } + } + }) +} + +fn capabilities() -> Value { + json!({ + "contract": 2, + "entities": sync::ENTITIES, + "attachments": {"maxBytes": attachments::MAX_BYTES, "types": attachments::TYPES}, + "ledgers": true + }) +} + +fn put_attachment( + state: &State, + space: &sync::Space, + sha: &str, + content_type: &str, + body: &[u8], +) -> Response { + use attachments::PutError; + match attachments::put( + &state.db, + &state.attachments, + space, + sha, + content_type, + body, + ) { + Ok(created) => json_response( + if created { 201 } else { 200 }, + json!({"sha256": sha, "size": body.len(), "contentType": attachments::media_type(content_type)}), + ), + Err(PutError::HashMismatch) => error( + 400, + "hash_mismatch", + "SHA-256 of the body does not match the path", + ), + Err(PutError::UnsupportedMediaType) => error( + 415, + "unsupported_media_type", + "Content-Type is not an accepted attachment type", + ), + Err(PutError::PayloadTooLarge) => error( + 413, + "payload_too_large", + "Attachment exceeds the maximum size", + ), + Err(PutError::Internal) => error(500, "internal_error", "Could not store attachment"), + } +} + +fn get_attachment(state: &State, space: &sync::Space, sha: &str) -> Response { + match attachments::get(&state.db, &state.attachments, space, sha) { + Ok(Some((bytes, media))) => { + let mut r = empty(200); + if let Ok(v) = HeaderValue::from_str(&media) { + r.headers_mut().insert("Content-Type", v); + } + r.headers_mut().insert( + "Content-Length", + HeaderValue::from_str(&bytes.len().to_string()).unwrap(), + ); + // Never render user files inline as part of the web client origin. + r.headers_mut().insert( + "Content-Disposition", + HeaderValue::from_static("attachment"), + ); + *r.body_mut() = Body::from(bytes); + r + } + Ok(None) => error(404, "not_found", "Attachment not found"), + Err(()) => error(500, "internal_error", "Could not read attachment"), + } +} + #[derive(Default)] struct Input { email: String, @@ -868,6 +1130,10 @@ mod tests { refresh_ttl: chrono::TimeDelta::days(30), allow_registration: true, cors_origins: vec!["https://allowed.example".into()], + attachments_dir: std::env::temp_dir().join(format!( + "balance-attachments-{}", + auth::identifier().unwrap() + )), }, Mutex::new(crate::database::open(":memory:").unwrap()), ) @@ -1010,3 +1276,6 @@ mod tests { ); } } + +#[cfg(test)] +mod v2_tests; diff --git a/src/server/v2_tests.rs b/src/server/v2_tests.rs new file mode 100644 index 0000000..a191172 --- /dev/null +++ b/src/server/v2_tests.rs @@ -0,0 +1,892 @@ +//! Endpoint tests for contract v2: capabilities, ledgers and attachments. + +use super::*; +use sha2::{Digest, Sha256}; + +struct Env { + state: State, + dir: std::path::PathBuf, +} + +impl Drop for Env { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.dir); + } +} + +fn env() -> Env { + let dir = std::env::temp_dir().join(format!("balance-v2-test-{}", auth::identifier().unwrap())); + let state = State::new( + Config { + address: "127.0.0.1:0".into(), + database_path: ":memory:".into(), + jwt_secret: "test secret".into(), + access_ttl: chrono::TimeDelta::minutes(15), + refresh_ttl: chrono::TimeDelta::days(30), + allow_registration: true, + cors_origins: vec!["https://allowed.example".into()], + attachments_dir: dir.clone(), + }, + Mutex::new(crate::database::open(":memory:").unwrap()), + ); + { + let db = state.db.lock().unwrap(); + for user in ["owner", "member", "stranger"] { + db.execute( + "INSERT INTO users VALUES(?,?,'hash','now')", + params![user, format!("{user}@example.com")], + ) + .unwrap(); + } + } + Env { state, dir } +} + +fn auth_headers(env: &Env, user: &str, ledger: Option<&str>) -> HeaderMap { + let (token, _) = env + .state + .tokens + .access_token(user, &format!("{user}@example.com")) + .unwrap(); + let mut headers = HeaderMap::new(); + headers.insert( + "Authorization", + HeaderValue::from_str(&format!("Bearer {token}")).unwrap(), + ); + if let Some(ledger) = ledger { + headers.insert("X-Balance-Ledger", HeaderValue::from_str(ledger).unwrap()); + } + headers +} + +fn call( + env: &Env, + method: &str, + path: &str, + user: Option<&str>, + ledger: Option<&str>, + body: &[u8], +) -> (u16, Value) { + let headers = match user { + Some(user) => auth_headers(env, user, ledger), + None => HeaderMap::new(), + }; + request(env, method, path, &headers, body) +} + +fn request(env: &Env, method: &str, path: &str, headers: &HeaderMap, body: &[u8]) -> (u16, Value) { + let r = dispatch( + &env.state, + &Method::from_bytes(method.as_bytes()).unwrap(), + path, + headers, + Ok(body), + "test-host", + ); + let status = r.status().as_u16(); + let bytes = futures_body(r); + ( + status, + serde_json::from_slice(&bytes).unwrap_or(Value::Null), + ) +} + +fn futures_body(r: Response) -> Vec { + tokio::runtime::Builder::new_current_thread() + .build() + .unwrap() + .block_on(axum::body::to_bytes(r.into_body(), usize::MAX)) + .unwrap() + .to_vec() +} + +fn code_of(v: &Value) -> &str { + v["error"]["code"].as_str().unwrap_or("") +} + +fn create_ledger(env: &Env) -> String { + let (status, v) = call( + env, + "POST", + "/v1/ledgers", + Some("owner"), + None, + br#"{"name":" Family "}"#, + ); + assert_eq!(status, 201); + v["id"].as_str().unwrap().to_owned() +} + +fn join_member(env: &Env, ledger: &str) { + let (_, invite) = call( + env, + "POST", + &format!("/v1/ledgers/{ledger}/invites"), + Some("owner"), + None, + b"", + ); + let body = json!({"code": invite["code"]}).to_string(); + let (status, _) = call( + env, + "POST", + "/v1/ledgers/join", + Some("member"), + None, + body.as_bytes(), + ); + assert_eq!(status, 200); +} + +fn sha(bytes: &[u8]) -> String { + Sha256::digest(bytes) + .iter() + .map(|b| format!("{b:02x}")) + .collect() +} + +fn put( + env: &Env, + user: &str, + ledger: Option<&str>, + hash: &str, + ty: &str, + bytes: &[u8], +) -> (u16, Value) { + let mut headers = auth_headers(env, user, ledger); + headers.insert("Content-Type", HeaderValue::from_str(ty).unwrap()); + request( + env, + "PUT", + &format!("/v1/attachments/{hash}"), + &headers, + bytes, + ) +} + +fn get_raw(env: &Env, user: &str, ledger: Option<&str>, hash: &str) -> Response { + dispatch( + &env.state, + &Method::GET, + &format!("/v1/attachments/{hash}"), + &auth_headers(env, user, ledger), + Ok(b""), + "test-host", + ) +} + +fn sync_call(env: &Env, user: &str, ledger: Option<&str>, body: Value) -> (u16, Value) { + call( + env, + "POST", + "/v1/sync", + Some(user), + ledger, + body.to_string().as_bytes(), + ) +} + +fn attachment_change(id: &str, hash: &str, time: &str, deleted: bool) -> Value { + json!({"entity":"attachment","id":id,"updatedAt":time,"deleted":deleted, + "payload":{"transactionId":"t","sha256":hash,"contentType":"image/png","size":3,"name":"a.png","createdAt":time}}) +} + +#[test] +fn capabilities_are_public_and_match_contract() { + let env = env(); + let (status, v) = call(&env, "GET", "/v1/capabilities", None, None, b""); + assert_eq!(status, 200); + assert_eq!( + v, + json!({ + "contract": 2, + "entities": ["transaction","category","budget","account","transfer", + "recurring","goal","debt","rate","setting","attachment"], + "attachments": {"maxBytes": 10485760, + "types": ["image/jpeg","image/png","image/heic","application/pdf"]}, + "ledgers": true + }) + ); + let r = dispatch( + &env.state, + &Method::POST, + "/v1/capabilities", + &HeaderMap::new(), + Ok(b""), + "h", + ); + assert_eq!(r.status(), StatusCode::METHOD_NOT_ALLOWED); + assert_eq!(r.headers()["Allow"], "GET, HEAD"); +} + +#[test] +fn sync_accepts_every_v2_entity_and_still_rejects_unknown_ones() { + let env = env(); + let changes: Vec = sync::ENTITIES + .iter() + .map(|e| json!({"entity":e,"id":format!("{e}-1"),"updatedAt":"2026-01-01T00:00:00Z","payload":{"x":1}})) + .collect(); + let (status, v) = sync_call(&env, "owner", None, json!({ "changes": changes })); + assert_eq!(status, 200); + assert_eq!(v["changes"].as_array().unwrap().len(), 11); + let (status, v) = sync_call( + &env, + "owner", + None, + json!({"changes":[{"entity":"wallet","id":"w","updatedAt":"2026-01-01T00:00:00Z","payload":{}}]}), + ); + assert_eq!(status, 400); + assert_eq!(code_of(&v), "invalid_change"); +} + +#[test] +fn v2_endpoints_require_authorization() { + let env = env(); + for (method, path) in [ + ("GET", "/v1/ledgers"), + ("POST", "/v1/ledgers"), + ("POST", "/v1/ledgers/join"), + ("PATCH", "/v1/ledgers/x"), + ("DELETE", "/v1/ledgers/x"), + ("POST", "/v1/ledgers/x/invites"), + ("DELETE", "/v1/ledgers/x/members/y"), + ("GET", "/v1/attachments/x"), + ("PUT", "/v1/attachments/x"), + ] { + let (status, v) = call(&env, method, path, None, None, b"{}"); + assert_eq!(status, 401, "{method} {path}"); + assert_eq!(code_of(&v), "authorization_required"); + } + let (status, _) = call(&env, "PUT", "/v1/ledgers", Some("owner"), None, b""); + assert_eq!(status, 405); + let (status, _) = call(&env, "GET", "/v1/ledgers/join", Some("owner"), None, b""); + assert_eq!(status, 405); +} + +#[test] +fn ledger_lifecycle_roles_and_error_codes() { + let env = env(); + // Personal space is not listed. + let (status, v) = call(&env, "GET", "/v1/ledgers", Some("owner"), None, b""); + assert_eq!((status, v), (200, json!({"ledgers": []}))); + for bad in [&br#"{"name":" "}"#[..], br#"{}"#] { + let (status, v) = call(&env, "POST", "/v1/ledgers", Some("owner"), None, bad); + assert_eq!((status, code_of(&v)), (400, "invalid_name")); + } + let (status, v) = call(&env, "POST", "/v1/ledgers", Some("owner"), None, b"{"); + assert_eq!((status, code_of(&v)), (400, "invalid_json")); + + let id = create_ledger(&env); + let (_, v) = call(&env, "GET", "/v1/ledgers", Some("owner"), None, b""); + assert_eq!( + v, + json!({"ledgers":[{"id":id,"name":"Family","role":"owner", + "members":[{"userId":"owner","email":"owner@example.com","role":"owner"}]}]}) + ); + let ledger = format!("/v1/ledgers/{id}"); + let invites = format!("{ledger}/invites"); + + // Unknown ledger → 404, non-member → 403. + let (status, v) = call( + &env, + "PATCH", + "/v1/ledgers/missing", + Some("owner"), + None, + br#"{"name":"x"}"#, + ); + assert_eq!((status, code_of(&v)), (404, "not_found")); + let (status, v) = call( + &env, + "PATCH", + &ledger, + Some("stranger"), + None, + br#"{"name":"x"}"#, + ); + assert_eq!((status, code_of(&v)), (403, "forbidden")); + let (status, v) = call(&env, "POST", &invites, Some("stranger"), None, b""); + assert_eq!((status, code_of(&v)), (403, "forbidden")); + + let (status, v) = call( + &env, + "PATCH", + &ledger, + Some("owner"), + None, + br#"{"name":"Home"}"#, + ); + assert_eq!((status, v["name"].as_str()), (200, Some("Home"))); + + // Invites: 10 characters, 7 days, hashed at rest, single use. + let (status, invite) = call(&env, "POST", &invites, Some("owner"), None, b""); + assert_eq!(status, 201); + let code = invite["code"].as_str().unwrap().to_owned(); + assert_eq!(code.len(), 10); + let expires = DateTime::parse_from_rfc3339(invite["expiresAt"].as_str().unwrap()).unwrap(); + let days = (expires.with_timezone(&Utc) - Utc::now()).num_hours(); + assert!((167..=168).contains(&days), "{days}"); + { + let db = env.state.db.lock().unwrap(); + let stored: String = db + .query_row("SELECT code_hash FROM ledger_invites", [], |r| r.get(0)) + .unwrap(); + assert_ne!(stored, code); + assert_eq!(stored, auth::token_hash(&code)); + } + for bad in [ + json!({"code":"AAAAAAAAAA"}), + json!({"code":"short"}), + json!({}), + ] { + let (status, v) = call( + &env, + "POST", + "/v1/ledgers/join", + Some("member"), + None, + bad.to_string().as_bytes(), + ); + assert_eq!((status, code_of(&v)), (400, "invalid_code")); + } + let lower = json!({"code": code.to_lowercase()}).to_string(); + let (status, v) = call( + &env, + "POST", + "/v1/ledgers/join", + Some("member"), + None, + lower.as_bytes(), + ); + assert_eq!(status, 200); + assert_eq!(v["role"], "member"); + assert_eq!(v["members"].as_array().unwrap().len(), 2); + let (status, v) = call( + &env, + "POST", + "/v1/ledgers/join", + Some("stranger"), + None, + lower.as_bytes(), + ); + assert_eq!((status, code_of(&v)), (400, "invalid_code")); + + // Expired codes are rejected. + let (_, expired) = call(&env, "POST", &invites, Some("owner"), None, b""); + env.state + .db + .lock() + .unwrap() + .execute( + "UPDATE ledger_invites SET expires_at='2000-01-01T00:00:00.000000000Z' WHERE code_hash=?", + [auth::token_hash(expired["code"].as_str().unwrap())], + ) + .unwrap(); + let body = json!({"code": expired["code"]}).to_string(); + let (status, v) = call( + &env, + "POST", + "/v1/ledgers/join", + Some("stranger"), + None, + body.as_bytes(), + ); + assert_eq!((status, code_of(&v)), (400, "invalid_code")); + + // Members may not administer the ledger. + let (status, v) = call( + &env, + "PATCH", + &ledger, + Some("member"), + None, + br#"{"name":"x"}"#, + ); + assert_eq!((status, code_of(&v)), (403, "forbidden")); + let (status, v) = call(&env, "POST", &invites, Some("member"), None, b""); + assert_eq!((status, code_of(&v)), (403, "forbidden")); + let (status, v) = call(&env, "DELETE", &ledger, Some("member"), None, b""); + assert_eq!((status, code_of(&v)), (403, "forbidden")); + let (status, v) = call( + &env, + "DELETE", + &format!("{ledger}/members/owner"), + Some("member"), + None, + b"", + ); + assert_eq!((status, code_of(&v)), (403, "forbidden")); + let (_, v) = call(&env, "GET", "/v1/ledgers", Some("member"), None, b""); + assert_eq!(v["ledgers"][0]["role"], "member"); + + // The owner cannot leave; unknown members are 404. + let (status, v) = call( + &env, + "DELETE", + &format!("{ledger}/members/owner"), + Some("owner"), + None, + b"", + ); + assert_eq!((status, code_of(&v)), (409, "owner_cannot_leave")); + let (status, v) = call( + &env, + "DELETE", + &format!("{ledger}/members/stranger"), + Some("owner"), + None, + b"", + ); + assert_eq!((status, code_of(&v)), (404, "not_found")); + + // A member leaves; the owner removes a member. + let (status, _) = call( + &env, + "DELETE", + &format!("{ledger}/members/member"), + Some("member"), + None, + b"", + ); + assert_eq!(status, 204); + let (_, v) = call(&env, "GET", "/v1/ledgers", Some("member"), None, b""); + assert_eq!(v, json!({"ledgers": []})); + join_member(&env, &id); + let (status, _) = call( + &env, + "DELETE", + &format!("{ledger}/members/member"), + Some("owner"), + None, + b"", + ); + assert_eq!(status, 204); + let (status, v) = sync_call(&env, "member", Some(&id), json!({})); + assert_eq!((status, code_of(&v)), (403, "forbidden")); + + let (status, _) = call(&env, "DELETE", &ledger, Some("owner"), None, b""); + assert_eq!(status, 204); + let (status, v) = call(&env, "DELETE", &ledger, Some("owner"), None, b""); + assert_eq!((status, code_of(&v)), (404, "not_found")); + let db = env.state.db.lock().unwrap(); + for table in ["ledgers", "ledger_members", "ledger_invites"] { + let n: i64 = db + .query_row(&format!("SELECT count(*) FROM {table}"), [], |r| r.get(0)) + .unwrap(); + assert_eq!(n, 0, "{table}"); + } +} + +#[test] +fn joining_twice_keeps_the_code_and_the_role() { + let env = env(); + let id = create_ledger(&env); + let (_, invite) = call( + &env, + "POST", + &format!("/v1/ledgers/{id}/invites"), + Some("owner"), + None, + b"", + ); + let body = json!({"code": invite["code"]}).to_string(); + let (status, v) = call( + &env, + "POST", + "/v1/ledgers/join", + Some("owner"), + None, + body.as_bytes(), + ); + assert_eq!((status, v["role"].as_str()), (200, Some("owner"))); + let (status, v) = call( + &env, + "POST", + "/v1/ledgers/join", + Some("member"), + None, + body.as_bytes(), + ); + assert_eq!((status, v["role"].as_str()), (200, Some("member"))); +} + +#[test] +fn join_attempts_are_rate_limited() { + let env = env(); + for _ in 0..30 { + let (status, _) = call( + &env, + "POST", + "/v1/ledgers/join", + Some("member"), + None, + br#"{"code":"x"}"#, + ); + assert_eq!(status, 400); + } + let (status, v) = call( + &env, + "POST", + "/v1/ledgers/join", + Some("member"), + None, + br#"{"code":"x"}"#, + ); + assert_eq!((status, code_of(&v)), (429, "rate_limited")); +} + +#[test] +fn ledger_sync_is_shared_isolated_and_has_its_own_cursor() { + let env = env(); + // Personal data first, so personal and ledger sequences differ. + let personal = json!({"changes":[{"entity":"category","id":"p","updatedAt":"2026-01-01T00:00:00Z","payload":{"name":"Mine"}}]}); + let (_, v) = sync_call(&env, "owner", None, personal.clone()); + assert_eq!(v["cursor"], 1); + let (_, v) = sync_call(&env, "owner", None, personal); + assert_eq!(v["cursor"], 1); + + let id = create_ledger(&env); + join_member(&env, &id); + let shared = json!({"changes":[{"entity":"transaction","id":"t","updatedAt":"2026-01-02T00:00:00Z","payload":{"amount":10}}]}); + let (status, v) = sync_call(&env, "owner", Some(&id), shared); + assert_eq!(status, 200); + assert_eq!(v["cursor"], 1); + assert_eq!(v["changes"][0]["id"], "t"); + + // The member sees the ledger record and nothing personal of the owner. + let (_, v) = sync_call(&env, "member", Some(&id), json!({})); + assert_eq!(v["changes"].as_array().unwrap().len(), 1); + assert_eq!(v["changes"][0]["entity"], "transaction"); + // Last write wins among members. + let edit = json!({"cursor":1,"changes":[{"entity":"transaction","id":"t","updatedAt":"2026-01-03T00:00:00Z","payload":{"amount":20}}]}); + let (_, v) = sync_call(&env, "member", Some(&id), edit); + assert_eq!(v["changes"][0]["version"], 2); + assert_eq!(v["cursor"], 2); + let (_, v) = sync_call(&env, "owner", Some(&id), json!({"cursor":1})); + assert_eq!(v["changes"][0]["payload"]["amount"], 20); + + // Personal spaces stay untouched. + let (_, v) = sync_call(&env, "member", None, json!({})); + assert_eq!( + (v["cursor"].clone(), v["changes"].clone()), + (json!(0), json!([])) + ); + let (_, v) = sync_call(&env, "owner", None, json!({})); + assert_eq!(v["changes"].as_array().unwrap().len(), 1); + assert_eq!(v["changes"][0]["id"], "p"); + // An empty header means the personal space. + let (_, v) = sync_call(&env, "owner", Some(" "), json!({})); + assert_eq!(v["changes"][0]["id"], "p"); + + let (status, v) = sync_call(&env, "stranger", Some(&id), json!({})); + assert_eq!((status, code_of(&v)), (403, "forbidden")); + let (status, v) = sync_call(&env, "owner", Some("missing"), json!({})); + assert_eq!((status, code_of(&v)), (404, "not_found")); + + // Deleting the ledger removes its records and changes. + call( + &env, + "DELETE", + &format!("/v1/ledgers/{id}"), + Some("owner"), + None, + b"", + ); + let db = env.state.db.lock().unwrap(); + for table in ["ledger_records", "ledger_changes"] { + let n: i64 = db + .query_row(&format!("SELECT count(*) FROM {table}"), [], |r| r.get(0)) + .unwrap(); + assert_eq!(n, 0, "{table}"); + } + let n: i64 = db + .query_row("SELECT count(*) FROM records", [], |r| r.get(0)) + .unwrap(); + assert_eq!(n, 1); +} + +#[test] +fn attachment_upload_validation_and_idempotency() { + let env = env(); + let bytes = b"\x89PNG fake"; + let hash = sha(bytes); + let (status, v) = put(&env, "owner", None, &hash, "image/png", bytes); + assert_eq!(status, 201); + assert_eq!(v["sha256"], hash.as_str()); + let (status, _) = put(&env, "owner", None, &hash, "image/png", bytes); + assert_eq!(status, 200); + + let (status, v) = put(&env, "owner", None, &sha(b"other"), "image/png", bytes); + assert_eq!((status, code_of(&v)), (400, "hash_mismatch")); + let (status, v) = put( + &env, + "owner", + None, + &hash.to_uppercase(), + "image/png", + bytes, + ); + assert_eq!((status, code_of(&v)), (400, "hash_mismatch")); + let (status, v) = put(&env, "owner", None, &hash, "image/gif", bytes); + assert_eq!((status, code_of(&v)), (415, "unsupported_media_type")); + let mut headers = auth_headers(&env, "owner", None); + let (status, v) = request( + &env, + "PUT", + &format!("/v1/attachments/{hash}"), + &headers, + bytes, + ); + assert_eq!((status, code_of(&v)), (415, "unsupported_media_type")); + headers.insert("Content-Type", HeaderValue::from_static("application/pdf")); + let large = vec![0u8; attachments::MAX_BYTES + 1]; + let (status, v) = request( + &env, + "PUT", + &format!("/v1/attachments/{}", sha(&large)), + &headers, + &large, + ); + assert_eq!((status, code_of(&v)), (413, "payload_too_large")); + let max = vec![1u8; attachments::MAX_BYTES]; + let (status, _) = request( + &env, + "PUT", + &format!("/v1/attachments/{}", sha(&max)), + &headers, + &max, + ); + assert_eq!(status, 201); + + let r = get_raw(&env, "owner", None, &hash); + assert_eq!(r.status(), StatusCode::OK); + assert_eq!(r.headers()["Content-Type"], "image/png"); + assert_eq!(r.headers()["Content-Disposition"], "attachment"); + assert_eq!(futures_body(r), bytes); + // Other users' spaces do not see the file. + assert_eq!( + get_raw(&env, "member", None, &hash).status(), + StatusCode::NOT_FOUND + ); + assert_eq!( + get_raw(&env, "owner", None, "nothex").status(), + StatusCode::NOT_FOUND + ); +} + +#[test] +fn attachments_are_per_ledger_and_check_membership() { + let env = env(); + let id = create_ledger(&env); + join_member(&env, &id); + let bytes = b"%PDF-1.7 receipt"; + let hash = sha(bytes); + assert_eq!( + put(&env, "member", Some(&id), &hash, "application/pdf", bytes).0, + 201 + ); + let r = get_raw(&env, "owner", Some(&id), &hash); + assert_eq!(r.status(), StatusCode::OK); + assert_eq!(futures_body(r), bytes); + // The same file in the personal space is stored separately. + assert_eq!( + get_raw(&env, "owner", None, &hash).status(), + StatusCode::NOT_FOUND + ); + assert_eq!( + put(&env, "owner", None, &hash, "application/pdf", bytes).0, + 201 + ); + + let (status, v) = put(&env, "stranger", Some(&id), &hash, "application/pdf", bytes); + assert_eq!((status, code_of(&v)), (403, "forbidden")); + assert_eq!( + get_raw(&env, "stranger", Some(&id), &hash).status(), + StatusCode::FORBIDDEN + ); + assert_eq!( + get_raw(&env, "owner", Some("missing"), &hash).status(), + StatusCode::NOT_FOUND + ); + + // Ledger deletion removes its blobs, the personal copy stays. + let ledger_dir = env.dir.join("ledgers").join(&id); + assert!(ledger_dir.is_dir()); + call( + &env, + "DELETE", + &format!("/v1/ledgers/{id}"), + Some("owner"), + None, + b"", + ); + assert!(!ledger_dir.exists()); + assert_eq!(get_raw(&env, "owner", None, &hash).status(), StatusCode::OK); + let n: i64 = env + .state + .db + .lock() + .unwrap() + .query_row("SELECT count(*) FROM attachment_blobs", [], |r| r.get(0)) + .unwrap(); + assert_eq!(n, 1); +} + +#[test] +fn blobs_are_collected_when_no_live_record_references_them() { + let env = env(); + let bytes = b"jpeg bytes"; + let hash = sha(bytes); + let pending = b"uploaded, not yet synced"; + let pending_hash = sha(pending); + assert_eq!(put(&env, "owner", None, &hash, "image/jpeg", bytes).0, 201); + assert_eq!( + put(&env, "owner", None, &pending_hash, "image/jpeg", pending).0, + 201 + ); + let file = env + .dir + .join("users") + .join("owner") + .join(&hash[..2]) + .join(&hash); + assert!(file.is_file()); + + let (status, _) = sync_call( + &env, + "owner", + None, + json!({"changes":[attachment_change("a1", &hash, "2026-01-01T00:00:00Z", false), + attachment_change("a2", &hash, "2026-01-01T00:00:00Z", false)]}), + ); + assert_eq!(status, 200); + // One of two referencing records is deleted: the blob stays. + sync_call( + &env, + "owner", + None, + json!({"changes":[attachment_change("a1", &hash, "2026-01-02T00:00:00Z", true)]}), + ); + assert_eq!(get_raw(&env, "owner", None, &hash).status(), StatusCode::OK); + // The last one is deleted: the blob is collected. + sync_call( + &env, + "owner", + None, + json!({"changes":[attachment_change("a2", &hash, "2026-01-02T00:00:00Z", true)]}), + ); + assert_eq!( + get_raw(&env, "owner", None, &hash).status(), + StatusCode::NOT_FOUND + ); + assert!(!file.exists()); + // A fresh upload whose record is not synced yet survives the grace period… + assert_eq!( + get_raw(&env, "owner", None, &pending_hash).status(), + StatusCode::OK + ); + // …but is collected once it is older than the grace period. + env.state + .db + .lock() + .unwrap() + .execute( + "UPDATE attachment_blobs SET uploaded_at='2000-01-01T00:00:00.000000000Z'", + [], + ) + .unwrap(); + sync_call( + &env, + "owner", + None, + json!({"changes":[attachment_change("a3", &hash, "2026-01-03T00:00:00Z", true)]}), + ); + assert_eq!( + get_raw(&env, "owner", None, &pending_hash).status(), + StatusCode::NOT_FOUND + ); + // Re-uploading a collected blob creates it again. + assert_eq!(put(&env, "owner", None, &hash, "image/jpeg", bytes).0, 201); +} + +#[tokio::test] +async fn large_uploads_pass_the_body_limit_only_when_authorized() { + let env = env(); + let state = Arc::new(State::new( + env.state.cfg.clone(), + Mutex::new(crate::database::open(":memory:").unwrap()), + )); + state + .db + .lock() + .unwrap() + .execute( + "INSERT INTO users VALUES('owner','owner@example.com','hash','now')", + [], + ) + .unwrap(); + let (token, _) = state + .tokens + .access_token("owner", "owner@example.com") + .unwrap(); + let bytes = vec![7u8; (2 << 20) + 10]; + let hash = sha(&bytes); + let r = handle( + ExtractState(state.clone()), + Request::builder() + .method("PUT") + .uri(format!("/v1/attachments/{hash}")) + .header("Authorization", format!("Bearer {token}")) + .header("Content-Type", "image/heic") + .body(Body::from(bytes.clone())) + .unwrap(), + ) + .await; + assert_eq!(r.status(), StatusCode::CREATED); + assert_eq!(r.headers()["Cache-Control"], "no-store"); + let r = handle( + ExtractState(state.clone()), + Request::builder() + .method("PUT") + .uri(format!("/v1/attachments/{hash}")) + .header("Content-Type", "image/heic") + .body(Body::from(bytes)) + .unwrap(), + ) + .await; + assert_eq!(r.status(), StatusCode::UNAUTHORIZED); +} + +#[tokio::test] +async fn cors_preflights_allow_v2_methods_and_ledger_header() { + let env = env(); + let state = Arc::new(State::new( + env.state.cfg.clone(), + Mutex::new(crate::database::open(":memory:").unwrap()), + )); + for (uri, request_method) in [("/v1/ledgers/x", None), ("/v1/sync", Some("POST"))] { + let mut builder = Request::builder() + .method("OPTIONS") + .uri(uri) + .header("Origin", "https://allowed.example"); + if let Some(m) = request_method { + builder = builder + .header("Access-Control-Request-Method", m) + .header("Access-Control-Request-Headers", "x-balance-ledger"); + } + let r = handle( + ExtractState(state.clone()), + builder.body(Body::empty()).unwrap(), + ) + .await; + assert_eq!(r.status(), StatusCode::NO_CONTENT); + assert_eq!( + r.headers()["Access-Control-Allow-Headers"], + "Authorization, Content-Type, X-Balance-Ledger" + ); + assert_eq!( + r.headers()["Access-Control-Allow-Methods"], + "GET, POST, PUT, PATCH, DELETE, OPTIONS" + ); + } +} diff --git a/src/sync.rs b/src/sync.rs index 95223e6..d75fdc2 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -259,11 +259,68 @@ fn failure(status: u16, code: &'static str, message: &str) -> SyncError { (status, code, message.to_owned()) } +/// Entities accepted by `/v1/sync` (contract v2, section 1). +pub const ENTITIES: [&str; 11] = [ + "transaction", + "category", + "budget", + "account", + "transfer", + "recurring", + "goal", + "debt", + "rate", + "setting", + "attachment", +]; + +/// A data space: the user's personal records (v1 tables) or a shared ledger. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum Space { + User(String), + Ledger(String), +} + +impl Space { + /// Records table, changes table and the column that selects the space. + pub fn tables(&self) -> (&'static str, &'static str, &'static str) { + match self { + Space::User(_) => ("records", "changes", "user_id"), + Space::Ledger(_) => ("ledger_records", "ledger_changes", "ledger_id"), + } + } + pub fn id(&self) -> &str { + match self { + Space::User(id) | Space::Ledger(id) => id, + } + } + /// Stable key of the space in `attachment_blobs`. + pub fn key(&self) -> String { + match self { + Space::User(id) => format!("user:{id}"), + Space::Ledger(id) => format!("ledger:{id}"), + } + } +} + +/// Synchronize the personal space of `user_id`. +#[cfg(test)] pub fn synchronize_wire( db: &mut Connection, user_id: &str, body: &[u8], ) -> Result, SyncError> { + synchronize_space(db, &Space::User(user_id.to_owned()), body).map(|(response, _)| response) +} + +/// Apply a sync request in `space`. Besides the response body, it returns the +/// attachment blobs that became garbage; their rows are already deleted in +/// the committed transaction and the caller removes the files. +pub fn synchronize_space( + db: &mut Connection, + space: &Space, + body: &[u8], +) -> Result<(Vec, Vec), SyncError> { let invalid_json = || failure(400, "invalid_json", "Request body is not valid JSON"); let body = &body[..body.len().min(2 << 20)]; // The original decoder reads the first JSON value, without requiring EOF. @@ -287,7 +344,9 @@ pub fn synchronize_wire( )); } for c in &request.changes { - let message = if !matches!(c.entity.as_str(), "transaction" | "category" | "budget") { + let message = if !ENTITIES.contains(&c.entity.as_str()) { + // The message is part of the recorded v1 reference responses + // (tools/compatibility_expected.txt), so it is kept unchanged. Some("entity must be transaction, category, or budget") } else if c.id.is_empty() || c.id.len() > 128 { Some("record id is invalid") @@ -313,12 +372,18 @@ pub fn synchronize_wire( let tx = db .transaction() .map_err(|_| failure(500, "sync_failed", "Could not begin synchronization"))?; + let touches_attachments = request.changes.iter().any(|c| c.entity == "attachment"); for c in request.changes { - apply(&tx, user_id, c) - .map_err(|_| failure(500, "sync_failed", "Could not apply changes"))?; + apply(&tx, space, c).map_err(|_| failure(500, "sync_failed", "Could not apply changes"))?; } + let garbage = if touches_attachments { + crate::attachments::collect_garbage(&tx, space, Utc::now()) + .map_err(|_| failure(500, "sync_failed", "Could not apply changes"))? + } else { + Vec::new() + }; let (changes, cursor, more) = if request.pull.unwrap_or(true) { - load(&tx, user_id, request.cursor) + load(&tx, space, request.cursor) .map_err(|_| failure(500, "sync_failed", "Could not load changes"))? } else { (Vec::new(), request.cursor, false) @@ -339,7 +404,7 @@ pub fn synchronize_wire( ) .as_bytes(), ); - Ok(response) + Ok((response, garbage)) } #[cfg(test)] @@ -351,10 +416,12 @@ fn synchronize(db: &mut Connection, user_id: &str, body: &[u8]) -> Result; -fn apply(tx: &Transaction<'_>, user: &str, c: Change) -> Result<(), InternalError> { +fn apply(tx: &Transaction<'_>, space: &Space, c: Change) -> Result<(), InternalError> { + let (records, changes, key) = space.tables(); + let user = space.id(); let current: Option<(String, i64)> = tx .query_row( - "SELECT updated_at,version FROM records WHERE user_id=? AND entity=? AND record_id=?", + &format!("SELECT updated_at,version FROM {records} WHERE {key}=? AND entity=? AND record_id=?"), params![user, c.entity, c.id], |r| Ok((r.get(0)?, r.get(1)?)), ) @@ -373,17 +440,19 @@ fn apply(tx: &Transaction<'_>, user: &str, c: Change) -> Result<(), InternalErro c.payload.as_deref() }; let time = timestamp(updated); - tx.execute("INSERT INTO records(user_id,entity,record_id,payload,updated_at,deleted,version) VALUES(?,?,?,CAST(? AS TEXT),?,?,?) ON CONFLICT(user_id,entity,record_id) DO UPDATE SET payload=excluded.payload,updated_at=excluded.updated_at,deleted=excluded.deleted,version=excluded.version", params![user,c.entity,c.id,payload,time,c.deleted,version])?; - tx.execute("INSERT INTO changes(user_id,entity,record_id,payload,updated_at,deleted,version) VALUES(?,?,?,CAST(? AS TEXT),?,?,?)", params![user,c.entity,c.id,payload,time,c.deleted,version])?; + tx.execute(&format!("INSERT INTO {records}({key},entity,record_id,payload,updated_at,deleted,version) VALUES(?,?,?,CAST(? AS TEXT),?,?,?) ON CONFLICT({key},entity,record_id) DO UPDATE SET payload=excluded.payload,updated_at=excluded.updated_at,deleted=excluded.deleted,version=excluded.version"), params![user,c.entity,c.id,payload,time,c.deleted,version])?; + tx.execute(&format!("INSERT INTO {changes}({key},entity,record_id,payload,updated_at,deleted,version) VALUES(?,?,?,CAST(? AS TEXT),?,?,?)"), params![user,c.entity,c.id,payload,time,c.deleted,version])?; Ok(()) } fn load( tx: &Transaction<'_>, - user: &str, + space: &Space, after: i64, ) -> Result<(Vec>, i64, bool), InternalError> { - let mut stmt = tx.prepare("SELECT sequence,entity,record_id,payload,updated_at,deleted,version FROM changes WHERE user_id=? AND sequence>? ORDER BY sequence ASC LIMIT ?")?; + let (_, changes, key) = space.tables(); + let user = space.id(); + let mut stmt = tx.prepare(&format!("SELECT sequence,entity,record_id,payload,updated_at,deleted,version FROM {changes} WHERE {key}=? AND sequence>? ORDER BY sequence ASC LIMIT ?"))?; let mut rows = stmt.query(params![user, after, 501])?; let mut result = Vec::new(); let mut cursor = after;