Compare commits
2
Commits
v0.3.0
...
multi-tenant
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
84323ab637 | ||
|
|
40f7f2fb5e |
Generated
+1
@@ -2574,6 +2574,7 @@ dependencies = [
|
|||||||
"time",
|
"time",
|
||||||
"tokio",
|
"tokio",
|
||||||
"tokio-util",
|
"tokio-util",
|
||||||
|
"tower",
|
||||||
"tower-http",
|
"tower-http",
|
||||||
"tracing",
|
"tracing",
|
||||||
"tracing-subscriber",
|
"tracing-subscriber",
|
||||||
|
|||||||
@@ -45,6 +45,7 @@ uuid = { version = "1", features = ["v4", "serde"] }
|
|||||||
|
|
||||||
[dev-dependencies]
|
[dev-dependencies]
|
||||||
cookie = { version = "0.18", features = ["signed"] }
|
cookie = { version = "0.18", features = ["signed"] }
|
||||||
|
tower = { version = "0.5", features = ["util"] }
|
||||||
|
|
||||||
[profile.release]
|
[profile.release]
|
||||||
lto = "thin"
|
lto = "thin"
|
||||||
|
|||||||
@@ -30,10 +30,13 @@ collect ratings and tags, and let clients download originals.
|
|||||||
`photos/<photo_id>/preview.jpg`, `photos/<photo_id>/thumb.jpg`. The bucket
|
`photos/<photo_id>/preview.jpg`, `photos/<photo_id>/thumb.jpg`. The bucket
|
||||||
stays fully private; all image traffic is streamed through the API with
|
stays fully private; all image traffic is streamed through the API with
|
||||||
auth checks (no bucket CORS or public access needed).
|
auth checks (no bucket CORS or public access needed).
|
||||||
- **Auth**: photographer signs in via any OIDC provider (authorization-code
|
- **Auth**: photographers sign in via any OIDC provider (authorization-code
|
||||||
flow + userinfo); only emails in `ALLOWED_EMAILS` may sign in, and sessions
|
flow + userinfo); only emails in `ALLOWED_EMAILS` may sign in, and sessions
|
||||||
are re-checked against the allowlist on every request, so removing an email
|
are re-checked against the allowlist on every request, so removing an email
|
||||||
revokes access immediately. Clients use unguessable share tokens, optionally
|
revokes access immediately. **Multi-tenant**: each allowed email is its own
|
||||||
|
workspace — albums, photos, and share links are owned per photographer and
|
||||||
|
invisible to the others (enforced via ownership-scoped data access and
|
||||||
|
covered by the tenant-isolation test matrix). Clients use unguessable share tokens, optionally
|
||||||
gated by an argon2-hashed password (10 wrong guesses lock the link for
|
gated by an argon2-hashed password (10 wrong guesses lock the link for
|
||||||
15 minutes).
|
15 minutes).
|
||||||
- **Frontend**: React + Vite SPA — justified gallery, lightbox with rating
|
- **Frontend**: React + Vite SPA — justified gallery, lightbox with rating
|
||||||
@@ -55,6 +58,13 @@ cargo run --bin worker # job worker (separate terminal, same env)
|
|||||||
cd frontend && npm install && npm run dev # UI on :5173, proxies /api
|
cd frontend && npm install && npm run dev # UI on :5173, proxies /api
|
||||||
```
|
```
|
||||||
|
|
||||||
|
Tests (the tenant-isolation matrix needs a disposable database):
|
||||||
|
|
||||||
|
```sh
|
||||||
|
createdb photos_test # or: docker compose exec postgres createdb -U photos photos_test
|
||||||
|
TEST_DATABASE_URL=postgres://photos:photos@localhost:5432/photos_test cargo test
|
||||||
|
```
|
||||||
|
|
||||||
Register the OIDC client with redirect URI `<PUBLIC_URL>/api/auth/callback`
|
Register the OIDC client with redirect URI `<PUBLIC_URL>/api/auth/callback`
|
||||||
(locally: `http://localhost:5173/api/auth/callback`). Any standard OIDC
|
(locally: `http://localhost:5173/api/auth/callback`). Any standard OIDC
|
||||||
provider works (Authentik, Keycloak, Zitadel, Dex, ...); the app uses
|
provider works (Authentik, Keycloak, Zitadel, Dex, ...); the app uses
|
||||||
@@ -158,8 +168,6 @@ Notes:
|
|||||||
- Full RAW develop fallback for files whose embedded preview is tiny
|
- Full RAW develop fallback for files whose embedded preview is tiny
|
||||||
(exceedingly rare on modern cameras; `darktable-cli` in the worker image
|
(exceedingly rare on modern cameras; `darktable-cli` in the worker image
|
||||||
would cover it).
|
would cover it).
|
||||||
- Multiple photographer accounts with separate libraries (any allowed email
|
|
||||||
sees everything).
|
|
||||||
- No S3 orphan sweeper: a crash in the narrow window between an upload's S3
|
- No S3 orphan sweeper: a crash in the narrow window between an upload's S3
|
||||||
put and its DB commit can leave an unreferenced original in the bucket
|
put and its DB commit can leave an unreferenced original in the bucket
|
||||||
(never data loss — just unclaimed storage).
|
(never data loss — just unclaimed storage).
|
||||||
|
|||||||
+8
-2
@@ -73,9 +73,15 @@ export function postDownload(url, ids = '') {
|
|||||||
form.remove()
|
form.remove()
|
||||||
}
|
}
|
||||||
|
|
||||||
// SHA-256 of a File, matching the server's content hash — used to skip
|
// Above this, hashing whole-file in memory (WebCrypto has no streaming digest)
|
||||||
// uploading bytes the album already has.
|
// risks OOM / the ~2GiB ArrayBuffer cap, so we skip the client dedup check and
|
||||||
|
// just upload — the server still dedups on arrival.
|
||||||
|
const CLIENT_HASH_LIMIT = 512 * 1024 * 1024
|
||||||
|
|
||||||
|
// SHA-256 of a File (lowercase hex), matching the server's content hash, or
|
||||||
|
// null when the file is too large to hash safely in the browser.
|
||||||
export async function sha256Hex(file) {
|
export async function sha256Hex(file) {
|
||||||
|
if (file.size > CLIENT_HASH_LIMIT) return null
|
||||||
const digest = await crypto.subtle.digest('SHA-256', await file.arrayBuffer())
|
const digest = await crypto.subtle.digest('SHA-256', await file.arrayBuffer())
|
||||||
return [...new Uint8Array(digest)].map((b) => b.toString(16).padStart(2, '0')).join('')
|
return [...new Uint8Array(digest)].map((b) => b.toString(16).padStart(2, '0')).join('')
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,4 +1,4 @@
|
|||||||
import { useEffect, useRef, useState } from 'react'
|
import { useCallback, useRef, useState } from 'react'
|
||||||
import { imgUrl } from '../api'
|
import { imgUrl } from '../api'
|
||||||
|
|
||||||
// True justified layout: pack photos greedily into rows at their real aspect
|
// True justified layout: pack photos greedily into rows at their real aspect
|
||||||
@@ -32,15 +32,19 @@ function layoutRows(photos, containerWidth, targetHeight, gap) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
export default function Gallery({ photos, onOpen, overlay, selected, onToggleSelect }) {
|
export default function Gallery({ photos, onOpen, overlay, selected, onToggleSelect }) {
|
||||||
const containerRef = useRef(null)
|
|
||||||
const [width, setWidth] = useState(0)
|
const [width, setWidth] = useState(0)
|
||||||
|
const observerRef = useRef(null)
|
||||||
|
|
||||||
useEffect(() => {
|
// Callback ref: (re)attaches the observer whenever the container node
|
||||||
const el = containerRef.current
|
// mounts. A plain mount-effect misses the case where Gallery first renders
|
||||||
if (!el) return
|
// empty (no container) and photos arrive later.
|
||||||
|
const containerRef = useCallback((node) => {
|
||||||
|
observerRef.current?.disconnect()
|
||||||
|
if (!node) return
|
||||||
const observer = new ResizeObserver((entries) => setWidth(entries[0].contentRect.width))
|
const observer = new ResizeObserver((entries) => setWidth(entries[0].contentRect.width))
|
||||||
observer.observe(el)
|
observer.observe(node)
|
||||||
return () => observer.disconnect()
|
observerRef.current = observer
|
||||||
|
setWidth(node.getBoundingClientRect().width)
|
||||||
}, [])
|
}, [])
|
||||||
|
|
||||||
if (photos.length === 0) return null
|
if (photos.length === 0) return null
|
||||||
|
|||||||
@@ -71,13 +71,16 @@ function UploadZone({ albumId, onUploaded }) {
|
|||||||
setQueue((q) => q.map((x) => (x.key === item.key ? { ...x, ...patch } : x)))
|
setQueue((q) => q.map((x) => (x.key === item.key ? { ...x, ...patch } : x)))
|
||||||
const transfer = async () => {
|
const transfer = async () => {
|
||||||
// Hash locally first: content the album already has is skipped
|
// Hash locally first: content the album already has is skipped
|
||||||
// without transferring a single byte.
|
// without transferring a single byte. Files too large to hash in the
|
||||||
|
// browser (null) fall straight through to a normal upload.
|
||||||
update({ status: 'checking' })
|
update({ status: 'checking' })
|
||||||
try {
|
try {
|
||||||
const hash = await sha256Hex(item.file)
|
const hash = await sha256Hex(item.file)
|
||||||
await api(`/api/albums/${albumId}/photos/by-hash/${hash}`)
|
if (hash) {
|
||||||
update({ status: 'skipped', progress: 1 })
|
await api(`/api/albums/${albumId}/photos/by-hash/${hash}`)
|
||||||
return
|
update({ status: 'skipped', progress: 1 })
|
||||||
|
return
|
||||||
|
}
|
||||||
} catch {
|
} catch {
|
||||||
// 404 (not there yet) or hashing unavailable — upload normally.
|
// 404 (not there yet) or hashing unavailable — upload normally.
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,27 @@
|
|||||||
|
-- Albums gain an owner: the tenancy root. Photos, shares, ratings and tags
|
||||||
|
-- all hang off albums, so this single column scopes everything.
|
||||||
|
alter table albums add column owner_id uuid references users(id);
|
||||||
|
|
||||||
|
-- Backfill: pre-tenancy albums had no owner. Assigning them to "the" original
|
||||||
|
-- photographer is only unambiguous when exactly one user exists. With several
|
||||||
|
-- (v0.1.0 let every allowed email share all albums) the correct owner is
|
||||||
|
-- unknowable, so refuse rather than silently transfer everyone's work to one
|
||||||
|
-- account — the operator must assign ownership manually before migrating.
|
||||||
|
do $$
|
||||||
|
declare
|
||||||
|
n_users int;
|
||||||
|
n_albums int;
|
||||||
|
begin
|
||||||
|
select count(*) into n_users from users;
|
||||||
|
select count(*) into n_albums from albums;
|
||||||
|
if n_albums > 0 and n_users <> 1 then
|
||||||
|
raise exception
|
||||||
|
'multi-tenant migration: % albums exist but there are % users (need exactly 1 to auto-assign ownership); set albums.owner_id manually first',
|
||||||
|
n_albums, n_users;
|
||||||
|
end if;
|
||||||
|
update albums set owner_id = (select id from users order by created_at limit 1);
|
||||||
|
end $$;
|
||||||
|
|
||||||
|
alter table albums alter column owner_id set not null;
|
||||||
|
|
||||||
|
create index albums_owner_idx on albums (owner_id);
|
||||||
+7
-1
@@ -114,11 +114,17 @@ pub(crate) fn base_cookie(name: &'static str, value: String, secure: bool) -> Co
|
|||||||
.build()
|
.build()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// The signed session-cookie payload, in one place so tests and the
|
||||||
|
/// mint_session dev tool can't drift from what user_from_jar parses.
|
||||||
|
pub fn session_payload(user_id: Uuid, email: &str, exp: i64) -> String {
|
||||||
|
format!("{user_id}|{exp}|{email}")
|
||||||
|
}
|
||||||
|
|
||||||
fn session_cookie(state: &AppState, user_id: Uuid, email: &str) -> Cookie<'static> {
|
fn session_cookie(state: &AppState, user_id: Uuid, email: &str) -> Cookie<'static> {
|
||||||
let exp = Utc::now().timestamp() + SESSION_DAYS * 86400;
|
let exp = Utc::now().timestamp() + SESSION_DAYS * 86400;
|
||||||
let mut cookie = base_cookie(
|
let mut cookie = base_cookie(
|
||||||
SESSION_COOKIE,
|
SESSION_COOKIE,
|
||||||
format!("{user_id}|{exp}|{email}"),
|
session_payload(user_id, email, exp),
|
||||||
state.config.cookie_secure(),
|
state.config.cookie_secure(),
|
||||||
);
|
);
|
||||||
cookie.set_max_age(time::Duration::days(SESSION_DAYS));
|
cookie.set_max_age(time::Duration::days(SESSION_DAYS));
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ pub mod error;
|
|||||||
pub mod imaging;
|
pub mod imaging;
|
||||||
pub mod jobs;
|
pub mod jobs;
|
||||||
pub mod models;
|
pub mod models;
|
||||||
|
pub mod owned;
|
||||||
pub mod routes;
|
pub mod routes;
|
||||||
pub mod s3;
|
pub mod s3;
|
||||||
pub mod state;
|
pub mod state;
|
||||||
|
|||||||
@@ -96,6 +96,8 @@ impl JobStatus {
|
|||||||
#[derive(Debug, Clone, sqlx::FromRow, Serialize)]
|
#[derive(Debug, Clone, sqlx::FromRow, Serialize)]
|
||||||
pub struct Album {
|
pub struct Album {
|
||||||
pub id: Uuid,
|
pub id: Uuid,
|
||||||
|
#[serde(skip_serializing)]
|
||||||
|
pub owner_id: Uuid,
|
||||||
pub name: String,
|
pub name: String,
|
||||||
pub description: String,
|
pub description: String,
|
||||||
pub created_at: DateTime<Utc>,
|
pub created_at: DateTime<Utc>,
|
||||||
|
|||||||
@@ -0,0 +1,46 @@
|
|||||||
|
//! Tenant-scoped data access. These are the ONLY functions admin handlers may
|
||||||
|
//! use to fetch albums, photos, or shares — every one joins ownership, so a
|
||||||
|
//! handler cannot accidentally reach across tenants. Another user's resource
|
||||||
|
//! is indistinguishable from a nonexistent one (404).
|
||||||
|
|
||||||
|
use uuid::Uuid;
|
||||||
|
|
||||||
|
use crate::error::ApiError;
|
||||||
|
use crate::models::{Album, Photo, Share};
|
||||||
|
use crate::state::AppState;
|
||||||
|
|
||||||
|
pub async fn album(state: &AppState, album_id: Uuid, owner: Uuid) -> Result<Album, ApiError> {
|
||||||
|
let album: Option<Album> =
|
||||||
|
sqlx::query_as("select * from albums where id = $1 and owner_id = $2")
|
||||||
|
.bind(album_id)
|
||||||
|
.bind(owner)
|
||||||
|
.fetch_optional(&state.db)
|
||||||
|
.await?;
|
||||||
|
album.ok_or_else(ApiError::not_found)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn photo(state: &AppState, photo_id: Uuid, owner: Uuid) -> Result<Photo, ApiError> {
|
||||||
|
let photo: Option<Photo> = sqlx::query_as(
|
||||||
|
"select p.* from photos p
|
||||||
|
join albums a on a.id = p.album_id
|
||||||
|
where p.id = $1 and a.owner_id = $2",
|
||||||
|
)
|
||||||
|
.bind(photo_id)
|
||||||
|
.bind(owner)
|
||||||
|
.fetch_optional(&state.db)
|
||||||
|
.await?;
|
||||||
|
photo.ok_or_else(ApiError::not_found)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn share(state: &AppState, share_id: Uuid, owner: Uuid) -> Result<Share, ApiError> {
|
||||||
|
let share: Option<Share> = sqlx::query_as(
|
||||||
|
"select s.* from shares s
|
||||||
|
join albums a on a.id = s.album_id
|
||||||
|
where s.id = $1 and a.owner_id = $2",
|
||||||
|
)
|
||||||
|
.bind(share_id)
|
||||||
|
.bind(owner)
|
||||||
|
.fetch_optional(&state.db)
|
||||||
|
.await?;
|
||||||
|
share.ok_or_else(ApiError::not_found)
|
||||||
|
}
|
||||||
+26
-15
@@ -6,8 +6,10 @@ use chrono::{DateTime, Utc};
|
|||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
use uuid::Uuid;
|
use uuid::Uuid;
|
||||||
|
|
||||||
|
use crate::auth::AuthUser;
|
||||||
use crate::error::{ApiError, ApiResult};
|
use crate::error::{ApiError, ApiResult};
|
||||||
use crate::models::{Album, JobKind, Photo, PhotoStatus};
|
use crate::models::{Album, JobKind, Photo, PhotoStatus};
|
||||||
|
use crate::owned;
|
||||||
use crate::state::AppState;
|
use crate::state::AppState;
|
||||||
|
|
||||||
#[derive(Serialize, sqlx::FromRow)]
|
#[derive(Serialize, sqlx::FromRow)]
|
||||||
@@ -23,6 +25,7 @@ pub struct AlbumListItem {
|
|||||||
|
|
||||||
pub async fn list(
|
pub async fn list(
|
||||||
State(state): State<AppState>,
|
State(state): State<AppState>,
|
||||||
|
user: AuthUser,
|
||||||
) -> ApiResult<Json<Vec<AlbumListItem>>> {
|
) -> ApiResult<Json<Vec<AlbumListItem>>> {
|
||||||
let albums: Vec<AlbumListItem> = sqlx::query_as(
|
let albums: Vec<AlbumListItem> = sqlx::query_as(
|
||||||
"select a.id, a.name, a.description, a.created_at,
|
"select a.id, a.name, a.description, a.created_at,
|
||||||
@@ -35,9 +38,11 @@ pub async fn list(
|
|||||||
order by coalesce(p.taken_at, p.created_at), p.filename
|
order by coalesce(p.taken_at, p.created_at), p.filename
|
||||||
limit 1
|
limit 1
|
||||||
) c on true
|
) c on true
|
||||||
|
where a.owner_id = $2
|
||||||
order by a.created_at desc",
|
order by a.created_at desc",
|
||||||
)
|
)
|
||||||
.bind(PhotoStatus::Ready.as_str())
|
.bind(PhotoStatus::Ready.as_str())
|
||||||
|
.bind(user.id)
|
||||||
.fetch_all(&state.db)
|
.fetch_all(&state.db)
|
||||||
.await?;
|
.await?;
|
||||||
Ok(Json(albums))
|
Ok(Json(albums))
|
||||||
@@ -52,18 +57,21 @@ pub struct CreateAlbum {
|
|||||||
|
|
||||||
pub async fn create(
|
pub async fn create(
|
||||||
State(state): State<AppState>,
|
State(state): State<AppState>,
|
||||||
|
user: AuthUser,
|
||||||
Json(body): Json<CreateAlbum>,
|
Json(body): Json<CreateAlbum>,
|
||||||
) -> ApiResult<Json<Album>> {
|
) -> ApiResult<Json<Album>> {
|
||||||
let name = body.name.trim();
|
let name = body.name.trim();
|
||||||
if name.is_empty() {
|
if name.is_empty() {
|
||||||
return Err(ApiError::bad_request("album name is required"));
|
return Err(ApiError::bad_request("album name is required"));
|
||||||
}
|
}
|
||||||
let album: Album =
|
let album: Album = sqlx::query_as(
|
||||||
sqlx::query_as("insert into albums (name, description) values ($1, $2) returning *")
|
"insert into albums (owner_id, name, description) values ($1, $2, $3) returning *",
|
||||||
.bind(name)
|
)
|
||||||
.bind(body.description.trim())
|
.bind(user.id)
|
||||||
.fetch_one(&state.db)
|
.bind(name)
|
||||||
.await?;
|
.bind(body.description.trim())
|
||||||
|
.fetch_one(&state.db)
|
||||||
|
.await?;
|
||||||
Ok(Json(album))
|
Ok(Json(album))
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -94,12 +102,10 @@ pub struct AlbumDetail {
|
|||||||
|
|
||||||
pub async fn get_one(
|
pub async fn get_one(
|
||||||
State(state): State<AppState>,
|
State(state): State<AppState>,
|
||||||
|
user: AuthUser,
|
||||||
Path(album_id): Path<Uuid>,
|
Path(album_id): Path<Uuid>,
|
||||||
) -> ApiResult<Json<AlbumDetail>> {
|
) -> ApiResult<Json<AlbumDetail>> {
|
||||||
let album: Album = sqlx::query_as("select * from albums where id = $1")
|
let album = owned::album(&state, album_id, user.id).await?;
|
||||||
.bind(album_id)
|
|
||||||
.fetch_one(&state.db)
|
|
||||||
.await?;
|
|
||||||
let photos: Vec<Photo> = sqlx::query_as(
|
let photos: Vec<Photo> = sqlx::query_as(
|
||||||
"select * from photos where album_id = $1
|
"select * from photos where album_id = $1
|
||||||
order by coalesce(taken_at, created_at), filename",
|
order by coalesce(taken_at, created_at), filename",
|
||||||
@@ -159,6 +165,7 @@ pub struct UpdateAlbum {
|
|||||||
|
|
||||||
pub async fn update(
|
pub async fn update(
|
||||||
State(state): State<AppState>,
|
State(state): State<AppState>,
|
||||||
|
user: AuthUser,
|
||||||
Path(album_id): Path<Uuid>,
|
Path(album_id): Path<Uuid>,
|
||||||
Json(body): Json<UpdateAlbum>,
|
Json(body): Json<UpdateAlbum>,
|
||||||
) -> ApiResult<Json<Album>> {
|
) -> ApiResult<Json<Album>> {
|
||||||
@@ -170,12 +177,13 @@ pub async fn update(
|
|||||||
let album: Album = sqlx::query_as(
|
let album: Album = sqlx::query_as(
|
||||||
"update albums
|
"update albums
|
||||||
set name = coalesce($2, name), description = coalesce($3, description)
|
set name = coalesce($2, name), description = coalesce($3, description)
|
||||||
where id = $1
|
where id = $1 and owner_id = $4
|
||||||
returning *",
|
returning *",
|
||||||
)
|
)
|
||||||
.bind(album_id)
|
.bind(album_id)
|
||||||
.bind(body.name.as_deref().map(str::trim))
|
.bind(body.name.as_deref().map(str::trim))
|
||||||
.bind(body.description.as_deref().map(str::trim))
|
.bind(body.description.as_deref().map(str::trim))
|
||||||
|
.bind(user.id)
|
||||||
.fetch_one(&state.db)
|
.fetch_one(&state.db)
|
||||||
.await?;
|
.await?;
|
||||||
Ok(Json(album))
|
Ok(Json(album))
|
||||||
@@ -183,16 +191,19 @@ pub async fn update(
|
|||||||
|
|
||||||
pub async fn delete(
|
pub async fn delete(
|
||||||
State(state): State<AppState>,
|
State(state): State<AppState>,
|
||||||
|
user: AuthUser,
|
||||||
Path(album_id): Path<Uuid>,
|
Path(album_id): Path<Uuid>,
|
||||||
) -> ApiResult<Json<serde_json::Value>> {
|
) -> ApiResult<Json<serde_json::Value>> {
|
||||||
let mut tx = state.db.begin().await?;
|
let mut tx = state.db.begin().await?;
|
||||||
// Lock the album row: concurrent uploads block on their FK check against
|
// Lock the album row: concurrent uploads block on their FK check against
|
||||||
// it, then fail once it's gone and clean up their own S3 objects — so no
|
// it, then fail once it's gone and clean up their own S3 objects — so no
|
||||||
// photo can slip in between the cleanup enqueue and the cascade delete.
|
// photo can slip in between the cleanup enqueue and the cascade delete.
|
||||||
let locked: Option<(Uuid,)> = sqlx::query_as("select id from albums where id = $1 for update")
|
let locked: Option<(Uuid,)> =
|
||||||
.bind(album_id)
|
sqlx::query_as("select id from albums where id = $1 and owner_id = $2 for update")
|
||||||
.fetch_optional(&mut *tx)
|
.bind(album_id)
|
||||||
.await?;
|
.bind(user.id)
|
||||||
|
.fetch_optional(&mut *tx)
|
||||||
|
.await?;
|
||||||
if locked.is_none() {
|
if locked.is_none() {
|
||||||
return Err(ApiError::not_found());
|
return Err(ApiError::not_found());
|
||||||
}
|
}
|
||||||
|
|||||||
+13
-2
@@ -28,8 +28,19 @@ async fn authorize_photo(
|
|||||||
.await?;
|
.await?;
|
||||||
let photo = photo.ok_or_else(ApiError::not_found)?;
|
let photo = photo.ok_or_else(ApiError::not_found)?;
|
||||||
|
|
||||||
if user_from_jar(state, jar).is_some() {
|
// Photographers may only reach their OWN photos (scoped by album owner);
|
||||||
return Ok(photo);
|
// otherwise fall through to share-cookie access.
|
||||||
|
if let Some(user) = user_from_jar(state, jar) {
|
||||||
|
let owns: Option<(Uuid,)> = sqlx::query_as(
|
||||||
|
"select a.id from albums a where a.id = $1 and a.owner_id = $2",
|
||||||
|
)
|
||||||
|
.bind(photo.album_id)
|
||||||
|
.bind(user.id)
|
||||||
|
.fetch_optional(&state.db)
|
||||||
|
.await?;
|
||||||
|
if owns.is_some() {
|
||||||
|
return Ok(photo);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
authorize_album_via_cookie(state, jar, photo.album_id, need_download).await?;
|
authorize_album_via_cookie(state, jar, photo.album_id, need_download).await?;
|
||||||
// Clients may only reach photos the share listing exposes.
|
// Clients may only reach photos the share listing exposes.
|
||||||
|
|||||||
+31
-9
@@ -9,9 +9,11 @@ use sha2::Digest;
|
|||||||
use tokio::io::AsyncWriteExt;
|
use tokio::io::AsyncWriteExt;
|
||||||
use uuid::Uuid;
|
use uuid::Uuid;
|
||||||
|
|
||||||
|
use crate::auth::AuthUser;
|
||||||
use crate::error::{ApiError, ApiResult};
|
use crate::error::{ApiError, ApiResult};
|
||||||
use crate::jobs;
|
use crate::jobs;
|
||||||
use crate::models::{JobKind, Photo, PhotoStatus};
|
use crate::models::{JobKind, Photo, PhotoStatus};
|
||||||
|
use crate::owned;
|
||||||
use crate::s3;
|
use crate::s3;
|
||||||
use crate::state::AppState;
|
use crate::state::AppState;
|
||||||
|
|
||||||
@@ -39,20 +41,34 @@ fn sanitize_filename(raw: &str) -> Result<String, ApiError> {
|
|||||||
Ok(cleaned.chars().take(150).collect())
|
Ok(cleaned.chars().take(150).collect())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// A dedup hit on a photo that previously failed processing means the user is
|
||||||
|
/// re-uploading to fix it — reset it and re-enqueue instead of handing back a
|
||||||
|
/// broken row that the UI would report as "already uploaded".
|
||||||
|
async fn heal_if_errored(state: &AppState, photo: Photo) -> ApiResult<Photo> {
|
||||||
|
if photo.status != PhotoStatus::Error {
|
||||||
|
return Ok(photo);
|
||||||
|
}
|
||||||
|
let mut tx = state.db.begin().await?;
|
||||||
|
let healed: Photo =
|
||||||
|
sqlx::query_as("update photos set status = $2, error = null where id = $1 returning *")
|
||||||
|
.bind(photo.id)
|
||||||
|
.bind(PhotoStatus::Uploaded.as_str())
|
||||||
|
.fetch_one(&mut *tx)
|
||||||
|
.await?;
|
||||||
|
jobs::ensure_process_photo(&mut tx, photo.id).await?;
|
||||||
|
tx.commit().await?;
|
||||||
|
Ok(healed)
|
||||||
|
}
|
||||||
|
|
||||||
pub async fn upload(
|
pub async fn upload(
|
||||||
State(state): State<AppState>,
|
State(state): State<AppState>,
|
||||||
|
user: AuthUser,
|
||||||
Path(album_id): Path<Uuid>,
|
Path(album_id): Path<Uuid>,
|
||||||
Query(query): Query<UploadQuery>,
|
Query(query): Query<UploadQuery>,
|
||||||
headers: HeaderMap,
|
headers: HeaderMap,
|
||||||
body: Body,
|
body: Body,
|
||||||
) -> ApiResult<Json<Photo>> {
|
) -> ApiResult<Json<Photo>> {
|
||||||
let album_exists: Option<(Uuid,)> = sqlx::query_as("select id from albums where id = $1")
|
owned::album(&state, album_id, user.id).await?;
|
||||||
.bind(album_id)
|
|
||||||
.fetch_optional(&state.db)
|
|
||||||
.await?;
|
|
||||||
if album_exists.is_none() {
|
|
||||||
return Err(ApiError::not_found());
|
|
||||||
}
|
|
||||||
|
|
||||||
let filename = sanitize_filename(&query.filename)?;
|
let filename = sanitize_filename(&query.filename)?;
|
||||||
let content_type = headers
|
let content_type = headers
|
||||||
@@ -97,7 +113,7 @@ pub async fn upload(
|
|||||||
.fetch_optional(&state.db)
|
.fetch_optional(&state.db)
|
||||||
.await?;
|
.await?;
|
||||||
if let Some(existing) = existing {
|
if let Some(existing) = existing {
|
||||||
return Ok(Json(existing));
|
return Ok(Json(heal_if_errored(&state, existing).await?));
|
||||||
}
|
}
|
||||||
|
|
||||||
// Upload to S3 first, then create the row and enqueue processing in one
|
// Upload to S3 first, then create the row and enqueue processing in one
|
||||||
@@ -156,7 +172,7 @@ pub async fn upload(
|
|||||||
.fetch_optional(&state.db)
|
.fetch_optional(&state.db)
|
||||||
.await?;
|
.await?;
|
||||||
if let Some(winner) = winner {
|
if let Some(winner) = winner {
|
||||||
return Ok(Json(winner));
|
return Ok(Json(heal_if_errored(&state, winner).await?));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return Err(e.into());
|
return Err(e.into());
|
||||||
@@ -178,8 +194,10 @@ pub async fn upload(
|
|||||||
/// content already exists in the album.
|
/// content already exists in the album.
|
||||||
pub async fn by_hash(
|
pub async fn by_hash(
|
||||||
State(state): State<AppState>,
|
State(state): State<AppState>,
|
||||||
|
user: AuthUser,
|
||||||
Path((album_id, sha256)): Path<(Uuid, String)>,
|
Path((album_id, sha256)): Path<(Uuid, String)>,
|
||||||
) -> ApiResult<Json<Photo>> {
|
) -> ApiResult<Json<Photo>> {
|
||||||
|
owned::album(&state, album_id, user.id).await?;
|
||||||
let photo: Option<Photo> =
|
let photo: Option<Photo> =
|
||||||
sqlx::query_as("select * from photos where album_id = $1 and sha256 = $2")
|
sqlx::query_as("select * from photos where album_id = $1 and sha256 = $2")
|
||||||
.bind(album_id)
|
.bind(album_id)
|
||||||
@@ -191,8 +209,10 @@ pub async fn by_hash(
|
|||||||
|
|
||||||
pub async fn delete(
|
pub async fn delete(
|
||||||
State(state): State<AppState>,
|
State(state): State<AppState>,
|
||||||
|
user: AuthUser,
|
||||||
Path(photo_id): Path<Uuid>,
|
Path(photo_id): Path<Uuid>,
|
||||||
) -> ApiResult<Json<serde_json::Value>> {
|
) -> ApiResult<Json<serde_json::Value>> {
|
||||||
|
owned::photo(&state, photo_id, user.id).await?;
|
||||||
let mut tx = state.db.begin().await?;
|
let mut tx = state.db.begin().await?;
|
||||||
let deleted = sqlx::query("delete from photos where id = $1")
|
let deleted = sqlx::query("delete from photos where id = $1")
|
||||||
.bind(photo_id)
|
.bind(photo_id)
|
||||||
@@ -213,8 +233,10 @@ pub async fn delete(
|
|||||||
|
|
||||||
pub async fn reprocess(
|
pub async fn reprocess(
|
||||||
State(state): State<AppState>,
|
State(state): State<AppState>,
|
||||||
|
user: AuthUser,
|
||||||
Path(photo_id): Path<Uuid>,
|
Path(photo_id): Path<Uuid>,
|
||||||
) -> ApiResult<Json<serde_json::Value>> {
|
) -> ApiResult<Json<serde_json::Value>> {
|
||||||
|
owned::photo(&state, photo_id, user.id).await?;
|
||||||
let mut tx = state.db.begin().await?;
|
let mut tx = state.db.begin().await?;
|
||||||
let updated = sqlx::query("update photos set status = $2, error = null where id = $1")
|
let updated = sqlx::query("update photos set status = $2, error = null where id = $1")
|
||||||
.bind(photo_id)
|
.bind(photo_id)
|
||||||
|
|||||||
+10
-8
@@ -7,8 +7,9 @@ use chrono::{DateTime, Utc};
|
|||||||
use serde::Deserialize;
|
use serde::Deserialize;
|
||||||
use uuid::Uuid;
|
use uuid::Uuid;
|
||||||
|
|
||||||
use crate::auth::random_token;
|
use crate::auth::{random_token, AuthUser};
|
||||||
use crate::error::{ApiError, ApiResult};
|
use crate::error::{ApiError, ApiResult};
|
||||||
|
use crate::owned;
|
||||||
use crate::state::AppState;
|
use crate::state::AppState;
|
||||||
|
|
||||||
#[derive(sqlx::FromRow)]
|
#[derive(sqlx::FromRow)]
|
||||||
@@ -48,8 +49,10 @@ const SHARE_COLUMNS: &str = "s.id, s.token, s.label, s.password_hash, s.allow_do
|
|||||||
|
|
||||||
pub async fn list(
|
pub async fn list(
|
||||||
State(state): State<AppState>,
|
State(state): State<AppState>,
|
||||||
|
user: AuthUser,
|
||||||
Path(album_id): Path<Uuid>,
|
Path(album_id): Path<Uuid>,
|
||||||
) -> ApiResult<Json<Vec<serde_json::Value>>> {
|
) -> ApiResult<Json<Vec<serde_json::Value>>> {
|
||||||
|
owned::album(&state, album_id, user.id).await?;
|
||||||
let rows: Vec<ShareAdminRow> = sqlx::query_as(&format!(
|
let rows: Vec<ShareAdminRow> = sqlx::query_as(&format!(
|
||||||
"select {SHARE_COLUMNS} from shares s where s.album_id = $1 order by s.created_at desc"
|
"select {SHARE_COLUMNS} from shares s where s.album_id = $1 order by s.created_at desc"
|
||||||
))
|
))
|
||||||
@@ -75,16 +78,11 @@ fn default_true() -> bool {
|
|||||||
|
|
||||||
pub async fn create(
|
pub async fn create(
|
||||||
State(state): State<AppState>,
|
State(state): State<AppState>,
|
||||||
|
user: AuthUser,
|
||||||
Path(album_id): Path<Uuid>,
|
Path(album_id): Path<Uuid>,
|
||||||
Json(body): Json<CreateShare>,
|
Json(body): Json<CreateShare>,
|
||||||
) -> ApiResult<Json<serde_json::Value>> {
|
) -> ApiResult<Json<serde_json::Value>> {
|
||||||
let album_exists: Option<(Uuid,)> = sqlx::query_as("select id from albums where id = $1")
|
owned::album(&state, album_id, user.id).await?;
|
||||||
.bind(album_id)
|
|
||||||
.fetch_optional(&state.db)
|
|
||||||
.await?;
|
|
||||||
if album_exists.is_none() {
|
|
||||||
return Err(ApiError::not_found());
|
|
||||||
}
|
|
||||||
|
|
||||||
let password_hash = match body.password.as_deref().map(str::trim) {
|
let password_hash = match body.password.as_deref().map(str::trim) {
|
||||||
Some(pw) if !pw.is_empty() => {
|
Some(pw) if !pw.is_empty() => {
|
||||||
@@ -131,8 +129,10 @@ pub async fn create(
|
|||||||
/// their way into the 15-minute lock).
|
/// their way into the 15-minute lock).
|
||||||
pub async fn reset_lock(
|
pub async fn reset_lock(
|
||||||
State(state): State<AppState>,
|
State(state): State<AppState>,
|
||||||
|
user: AuthUser,
|
||||||
Path(share_id): Path<Uuid>,
|
Path(share_id): Path<Uuid>,
|
||||||
) -> ApiResult<Json<serde_json::Value>> {
|
) -> ApiResult<Json<serde_json::Value>> {
|
||||||
|
owned::share(&state, share_id, user.id).await?;
|
||||||
let updated =
|
let updated =
|
||||||
sqlx::query("update shares set failed_attempts = 0, locked_until = null where id = $1")
|
sqlx::query("update shares set failed_attempts = 0, locked_until = null where id = $1")
|
||||||
.bind(share_id)
|
.bind(share_id)
|
||||||
@@ -146,8 +146,10 @@ pub async fn reset_lock(
|
|||||||
|
|
||||||
pub async fn delete(
|
pub async fn delete(
|
||||||
State(state): State<AppState>,
|
State(state): State<AppState>,
|
||||||
|
user: AuthUser,
|
||||||
Path(share_id): Path<Uuid>,
|
Path(share_id): Path<Uuid>,
|
||||||
) -> ApiResult<Json<serde_json::Value>> {
|
) -> ApiResult<Json<serde_json::Value>> {
|
||||||
|
owned::share(&state, share_id, user.id).await?;
|
||||||
let deleted = sqlx::query("delete from shares where id = $1")
|
let deleted = sqlx::query("delete from shares where id = $1")
|
||||||
.bind(share_id)
|
.bind(share_id)
|
||||||
.execute(&state.db)
|
.execute(&state.db)
|
||||||
|
|||||||
+46
-59
@@ -81,12 +81,13 @@ async fn album_name(state: &AppState, album_id: Uuid) -> Result<String, ApiError
|
|||||||
|
|
||||||
pub async fn album_zip(
|
pub async fn album_zip(
|
||||||
State(state): State<AppState>,
|
State(state): State<AppState>,
|
||||||
|
user: crate::auth::AuthUser,
|
||||||
Path(album_id): Path<Uuid>,
|
Path(album_id): Path<Uuid>,
|
||||||
Form(request): Form<ZipRequest>,
|
Form(request): Form<ZipRequest>,
|
||||||
) -> ApiResult<Response> {
|
) -> ApiResult<Response> {
|
||||||
let name = album_name(&state, album_id).await?;
|
let album = crate::owned::album(&state, album_id, user.id).await?;
|
||||||
let photos = ready_photos(&state, album_id, &parse_ids(&request.ids)?).await?;
|
let photos = ready_photos(&state, album_id, &parse_ids(&request.ids)?).await?;
|
||||||
stream_zip(state, photos, &name)
|
stream_zip(state, photos, &album.name)
|
||||||
}
|
}
|
||||||
|
|
||||||
pub async fn share_zip(
|
pub async fn share_zip(
|
||||||
@@ -136,9 +137,6 @@ struct Entry {
|
|||||||
offset: u64,
|
offset: u64,
|
||||||
dos_time: u16,
|
dos_time: u16,
|
||||||
dos_date: u16,
|
dos_date: u16,
|
||||||
/// Stored at upload/processing time; photos from before hashes existed
|
|
||||||
/// have None and take the slower spool path (which backfills it).
|
|
||||||
crc: Option<u32>,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
struct ZipPlan {
|
struct ZipPlan {
|
||||||
@@ -179,7 +177,6 @@ fn plan_zip(photos: &[Photo]) -> Result<ZipPlan, ApiError> {
|
|||||||
offset: entry_offset,
|
offset: entry_offset,
|
||||||
dos_time,
|
dos_time,
|
||||||
dos_date,
|
dos_date,
|
||||||
crc: photo.crc32.map(|v| v as u32),
|
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
let cd_offset = offset;
|
let cd_offset = offset;
|
||||||
@@ -281,23 +278,26 @@ fn stream_zip(state: AppState, photos: Vec<Photo>, album_name: &str) -> ApiResul
|
|||||||
.map_err(|e| anyhow::anyhow!("building response: {e}").into())
|
.map_err(|e| anyhow::anyhow!("building response: {e}").into())
|
||||||
}
|
}
|
||||||
|
|
||||||
enum Fetched {
|
struct Fetched {
|
||||||
/// CRC already known — the body streams straight into the response.
|
file: tokio::fs::File,
|
||||||
Direct(Box<aws_sdk_s3::operation::get_object::GetObjectOutput>),
|
crc: u32,
|
||||||
/// Pre-hash photo: spooled to a temp file to compute the CRC first.
|
/// sha256 hex — always computed so legacy photos (crc-only) get their
|
||||||
Spooled(tokio::fs::File, u32),
|
/// content hash backfilled, restoring upload dedup for them.
|
||||||
|
sha256: String,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Start fetching an original. With a known CRC this only opens the S3
|
/// Fetch an original and spool it to an anonymous temp file, computing crc32
|
||||||
/// response (the body is consumed later, straight into the zip stream);
|
/// and sha256 and verifying the byte count the zip plan promised. Spooling
|
||||||
/// otherwise the object is spooled to an anonymous temp file to compute the
|
/// (rather than streaming the live S3 body straight through) is deliberate:
|
||||||
/// CRC, verifying the byte count the zip plan promised.
|
/// the prefetched entry is fully read immediately, so no S3 connection sits
|
||||||
|
/// idle across the previous entry's (possibly slow) client stream — which S3
|
||||||
|
/// idle-timeouts would otherwise reset mid-download.
|
||||||
fn fetch_entry(
|
fn fetch_entry(
|
||||||
state: &AppState,
|
state: &AppState,
|
||||||
key: String,
|
key: String,
|
||||||
expected_size: u64,
|
expected_size: u64,
|
||||||
crc_known: bool,
|
|
||||||
) -> JoinHandle<anyhow::Result<Fetched>> {
|
) -> JoinHandle<anyhow::Result<Fetched>> {
|
||||||
|
use sha2::Digest;
|
||||||
let state = state.clone();
|
let state = state.clone();
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
let object = state
|
let object = state
|
||||||
@@ -308,12 +308,10 @@ fn fetch_entry(
|
|||||||
.send()
|
.send()
|
||||||
.await
|
.await
|
||||||
.map_err(|e| anyhow::anyhow!("fetching {key}: {e}"))?;
|
.map_err(|e| anyhow::anyhow!("fetching {key}: {e}"))?;
|
||||||
if crc_known {
|
|
||||||
return Ok(Fetched::Direct(Box::new(object)));
|
|
||||||
}
|
|
||||||
let mut file = tokio::fs::File::from_std(tempfile::tempfile()?);
|
let mut file = tokio::fs::File::from_std(tempfile::tempfile()?);
|
||||||
let mut reader = object.body.into_async_read();
|
let mut reader = object.body.into_async_read();
|
||||||
let mut hasher = crc32fast::Hasher::new();
|
let mut crc = crc32fast::Hasher::new();
|
||||||
|
let mut sha = sha2::Sha256::new();
|
||||||
let mut written: u64 = 0;
|
let mut written: u64 = 0;
|
||||||
let mut buf = vec![0u8; 128 * 1024];
|
let mut buf = vec![0u8; 128 * 1024];
|
||||||
loop {
|
loop {
|
||||||
@@ -321,7 +319,8 @@ fn fetch_entry(
|
|||||||
if n == 0 {
|
if n == 0 {
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
hasher.update(&buf[..n]);
|
crc.update(&buf[..n]);
|
||||||
|
sha.update(&buf[..n]);
|
||||||
file.write_all(&buf[..n]).await?;
|
file.write_all(&buf[..n]).await?;
|
||||||
written += n as u64;
|
written += n as u64;
|
||||||
}
|
}
|
||||||
@@ -331,7 +330,11 @@ fn fetch_entry(
|
|||||||
);
|
);
|
||||||
file.flush().await?;
|
file.flush().await?;
|
||||||
file.seek(std::io::SeekFrom::Start(0)).await?;
|
file.seek(std::io::SeekFrom::Start(0)).await?;
|
||||||
Ok(Fetched::Spooled(file, hasher.finalize()))
|
Ok(Fetched {
|
||||||
|
file,
|
||||||
|
crc: crc.finalize(),
|
||||||
|
sha256: hex::encode(sha.finalize()),
|
||||||
|
})
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -344,30 +347,23 @@ async fn write_zip(
|
|||||||
const FLAGS: u16 = 0x0800;
|
const FLAGS: u16 = 0x0800;
|
||||||
let mut crcs = Vec::with_capacity(plan.entries.len());
|
let mut crcs = Vec::with_capacity(plan.entries.len());
|
||||||
|
|
||||||
// Prefetch: start fetching the next object while streaming the current one.
|
// Prefetch: spool the next object to a temp file while streaming the
|
||||||
|
// current one — the prefetched body is drained immediately, never held
|
||||||
|
// open across the current entry's client stream.
|
||||||
let mut pending: Option<JoinHandle<anyhow::Result<Fetched>>> = None;
|
let mut pending: Option<JoinHandle<anyhow::Result<Fetched>>> = None;
|
||||||
for (i, entry) in plan.entries.iter().enumerate() {
|
for (i, entry) in plan.entries.iter().enumerate() {
|
||||||
let current = match pending.take() {
|
let current = match pending.take() {
|
||||||
Some(handle) => handle,
|
Some(handle) => handle,
|
||||||
None => fetch_entry(state, entry.s3_key.clone(), entry.size, entry.crc.is_some()),
|
None => fetch_entry(state, entry.s3_key.clone(), entry.size),
|
||||||
};
|
};
|
||||||
if let Some(next) = plan.entries.get(i + 1) {
|
if let Some(next) = plan.entries.get(i + 1) {
|
||||||
pending = Some(fetch_entry(
|
pending = Some(fetch_entry(state, next.s3_key.clone(), next.size));
|
||||||
state,
|
|
||||||
next.s3_key.clone(),
|
|
||||||
next.size,
|
|
||||||
next.crc.is_some(),
|
|
||||||
));
|
|
||||||
}
|
}
|
||||||
let fetched = current
|
let mut fetched = current
|
||||||
.await
|
.await
|
||||||
.map_err(|e| anyhow::anyhow!("fetch task failed: {e}"))?
|
.map_err(|e| anyhow::anyhow!("fetch task failed: {e}"))?
|
||||||
.map_err(|e| anyhow::anyhow!("photo {}: {e:#}", entry.photo_id))?;
|
.map_err(|e| anyhow::anyhow!("photo {}: {e:#}", entry.photo_id))?;
|
||||||
let crc = match &fetched {
|
crcs.push(fetched.crc);
|
||||||
Fetched::Direct(_) => entry.crc.expect("direct fetch implies known crc"),
|
|
||||||
Fetched::Spooled(_, crc) => *crc,
|
|
||||||
};
|
|
||||||
crcs.push(crc);
|
|
||||||
|
|
||||||
let mut lfh = Vec::with_capacity(30 + entry.name.len());
|
let mut lfh = Vec::with_capacity(30 + entry.name.len());
|
||||||
lfh.extend_from_slice(&0x04034b50u32.to_le_bytes());
|
lfh.extend_from_slice(&0x04034b50u32.to_le_bytes());
|
||||||
@@ -376,35 +372,26 @@ async fn write_zip(
|
|||||||
lfh.extend_from_slice(&0u16.to_le_bytes()); // method: stored
|
lfh.extend_from_slice(&0u16.to_le_bytes()); // method: stored
|
||||||
lfh.extend_from_slice(&entry.dos_time.to_le_bytes());
|
lfh.extend_from_slice(&entry.dos_time.to_le_bytes());
|
||||||
lfh.extend_from_slice(&entry.dos_date.to_le_bytes());
|
lfh.extend_from_slice(&entry.dos_date.to_le_bytes());
|
||||||
lfh.extend_from_slice(&crc.to_le_bytes());
|
lfh.extend_from_slice(&fetched.crc.to_le_bytes());
|
||||||
lfh.extend_from_slice(&(entry.size as u32).to_le_bytes()); // compressed
|
lfh.extend_from_slice(&(entry.size as u32).to_le_bytes()); // compressed
|
||||||
lfh.extend_from_slice(&(entry.size as u32).to_le_bytes()); // uncompressed
|
lfh.extend_from_slice(&(entry.size as u32).to_le_bytes()); // uncompressed
|
||||||
lfh.extend_from_slice(&(entry.name.len() as u16).to_le_bytes());
|
lfh.extend_from_slice(&(entry.name.len() as u16).to_le_bytes());
|
||||||
lfh.extend_from_slice(&0u16.to_le_bytes()); // extra len
|
lfh.extend_from_slice(&0u16.to_le_bytes()); // extra len
|
||||||
lfh.extend_from_slice(&entry.name);
|
lfh.extend_from_slice(&entry.name);
|
||||||
out.write_all(&lfh).await?;
|
out.write_all(&lfh).await?;
|
||||||
match fetched {
|
tokio::io::copy(&mut fetched.file, &mut out).await?;
|
||||||
Fetched::Direct(object) => {
|
|
||||||
let mut reader = object.body.into_async_read();
|
// Self-heal legacy photos: backfill both hashes so future zips and
|
||||||
let copied = tokio::io::copy(&mut reader, &mut out).await?;
|
// upload dedup both work for them.
|
||||||
anyhow::ensure!(
|
let _ = sqlx::query(
|
||||||
copied == entry.size,
|
"update photos set crc32 = coalesce(crc32, $2), sha256 = coalesce(sha256, $3)
|
||||||
"{} is {copied} bytes in s3 but {} in the database",
|
where id = $1",
|
||||||
entry.s3_key,
|
)
|
||||||
entry.size
|
.bind(entry.photo_id)
|
||||||
);
|
.bind(i64::from(fetched.crc))
|
||||||
}
|
.bind(&fetched.sha256)
|
||||||
Fetched::Spooled(mut file, crc) => {
|
.execute(&state.db)
|
||||||
tokio::io::copy(&mut file, &mut out).await?;
|
.await;
|
||||||
// Self-heal: store the freshly computed crc so the next
|
|
||||||
// download of this photo streams directly.
|
|
||||||
let _ = sqlx::query("update photos set crc32 = coalesce(crc32, $2) where id = $1")
|
|
||||||
.bind(entry.photo_id)
|
|
||||||
.bind(i64::from(crc))
|
|
||||||
.execute(&state.db)
|
|
||||||
.await;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Central directory.
|
// Central directory.
|
||||||
|
|||||||
@@ -0,0 +1,254 @@
|
|||||||
|
//! Tenant-isolation matrix: every admin endpoint must treat another user's
|
||||||
|
//! resources as nonexistent (404), and unauthenticated requests as 401.
|
||||||
|
//!
|
||||||
|
//! Needs a disposable Postgres database:
|
||||||
|
//! TEST_DATABASE_URL=postgres://photos:photos@localhost:5432/photos_test cargo test
|
||||||
|
//! Skips silently when TEST_DATABASE_URL is unset. S3 is never contacted —
|
||||||
|
//! the matrix stops at the ownership checks by construction.
|
||||||
|
|
||||||
|
use axum::body::Body;
|
||||||
|
use axum::http::{header, Request, StatusCode};
|
||||||
|
use cookie::{Cookie, CookieJar, Key};
|
||||||
|
use sha2::{Digest, Sha512};
|
||||||
|
use tower::ServiceExt;
|
||||||
|
use uuid::Uuid;
|
||||||
|
|
||||||
|
use photos::config::Config;
|
||||||
|
use photos::state::AppState;
|
||||||
|
|
||||||
|
const SECRET: &str = "test-secret-test-secret-test-secret-1234";
|
||||||
|
|
||||||
|
fn test_config(database_url: String) -> Config {
|
||||||
|
Config {
|
||||||
|
database_url,
|
||||||
|
bind_addr: "127.0.0.1:0".into(),
|
||||||
|
public_url: "http://localhost:8080".into(),
|
||||||
|
session_secret: SECRET.into(),
|
||||||
|
s3_bucket: "photos".into(),
|
||||||
|
// Unroutable on purpose: no test may reach S3.
|
||||||
|
s3_endpoint: Some("http://127.0.0.1:9".into()),
|
||||||
|
s3_region: "us-east-1".into(),
|
||||||
|
s3_access_key: "test".into(),
|
||||||
|
s3_secret_key: "test".into(),
|
||||||
|
s3_force_path_style: true,
|
||||||
|
oidc_issuer: "https://auth.invalid".into(),
|
||||||
|
oidc_client_id: "x".into(),
|
||||||
|
oidc_client_secret: "x".into(),
|
||||||
|
allowed_emails: vec!["a@test".into(), "b@test".into()],
|
||||||
|
static_dir: "frontend/dist".into(),
|
||||||
|
worker_concurrency: 1,
|
||||||
|
dev_autologin_email: None,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn session_for(user_id: Uuid, email: &str) -> String {
|
||||||
|
let key = Key::from(&Sha512::digest(SECRET.as_bytes()));
|
||||||
|
let exp = chrono::Utc::now().timestamp() + 3600;
|
||||||
|
// Use the production payload builder so the test can't drift from what
|
||||||
|
// user_from_jar parses.
|
||||||
|
let mut jar = CookieJar::new();
|
||||||
|
jar.signed_mut(&key).add(Cookie::new(
|
||||||
|
"photos_session",
|
||||||
|
photos::auth::session_payload(user_id, email, exp),
|
||||||
|
));
|
||||||
|
format!("photos_session={}", jar.get("photos_session").unwrap().value())
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn request(
|
||||||
|
router: &axum::Router,
|
||||||
|
method: &str,
|
||||||
|
path: &str,
|
||||||
|
cookie: Option<&str>,
|
||||||
|
json: Option<serde_json::Value>,
|
||||||
|
) -> (StatusCode, serde_json::Value) {
|
||||||
|
let mut builder = Request::builder().method(method).uri(path);
|
||||||
|
if let Some(cookie) = cookie {
|
||||||
|
builder = builder.header(header::COOKIE, cookie);
|
||||||
|
}
|
||||||
|
let body = match json {
|
||||||
|
Some(value) => {
|
||||||
|
builder = builder.header(header::CONTENT_TYPE, "application/json");
|
||||||
|
Body::from(value.to_string())
|
||||||
|
}
|
||||||
|
// Zip endpoints take a form; everything else ignores the body.
|
||||||
|
None if path.ends_with("/zip") => {
|
||||||
|
builder = builder.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded");
|
||||||
|
Body::from("ids=")
|
||||||
|
}
|
||||||
|
None => Body::empty(),
|
||||||
|
};
|
||||||
|
let response = router
|
||||||
|
.clone()
|
||||||
|
.oneshot(builder.body(body).unwrap())
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
let status = response.status();
|
||||||
|
let bytes = axum::body::to_bytes(response.into_body(), 1 << 20)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
let value = serde_json::from_slice(&bytes).unwrap_or(serde_json::Value::Null);
|
||||||
|
(status, value)
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn seed_user(state: &AppState, email: &str) -> Uuid {
|
||||||
|
let (id,): (Uuid,) = sqlx::query_as(
|
||||||
|
"insert into users (oidc_subject, email) values ($1, $2) returning id",
|
||||||
|
)
|
||||||
|
.bind(format!("test:{email}"))
|
||||||
|
.bind(email)
|
||||||
|
.fetch_one(&state.db)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
id
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn tenant_isolation_matrix() {
|
||||||
|
let Ok(database_url) = std::env::var("TEST_DATABASE_URL") else {
|
||||||
|
eprintln!("TEST_DATABASE_URL not set — skipping tenancy matrix");
|
||||||
|
return;
|
||||||
|
};
|
||||||
|
let state = AppState::new(test_config(database_url)).await.unwrap();
|
||||||
|
sqlx::query("truncate users, albums, photos, shares, ratings, tags, jobs cascade")
|
||||||
|
.execute(&state.db)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let alice = seed_user(&state, "a@test").await;
|
||||||
|
let bob = seed_user(&state, "b@test").await;
|
||||||
|
let cookie_a = session_for(alice, "a@test");
|
||||||
|
let cookie_b = session_for(bob, "b@test");
|
||||||
|
let router = photos::routes::router(&state).with_state(state.clone());
|
||||||
|
|
||||||
|
// Alice creates an album through the real API.
|
||||||
|
let (status, album) = request(
|
||||||
|
&router,
|
||||||
|
"POST",
|
||||||
|
"/api/albums",
|
||||||
|
Some(&cookie_a),
|
||||||
|
Some(serde_json::json!({ "name": "Alice's Wedding" })),
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
assert_eq!(status, StatusCode::OK);
|
||||||
|
let album_id = album["id"].as_str().unwrap().to_string();
|
||||||
|
|
||||||
|
// Seed a photo (kept non-ready so no code path reaches S3) and a share.
|
||||||
|
let photo_id = Uuid::new_v4();
|
||||||
|
sqlx::query(
|
||||||
|
"insert into photos (id, album_id, filename, content_type, size_bytes, sha256)
|
||||||
|
values ($1, $2::uuid, 'a.jpg', 'image/jpeg', 3, 'hash-a')",
|
||||||
|
)
|
||||||
|
.bind(photo_id)
|
||||||
|
.bind(&album_id)
|
||||||
|
.execute(&state.db)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
let share_id = Uuid::new_v4();
|
||||||
|
sqlx::query("insert into shares (id, album_id, token) values ($1, $2::uuid, 'tenanttesttoken123456789')")
|
||||||
|
.bind(share_id)
|
||||||
|
.bind(&album_id)
|
||||||
|
.execute(&state.db)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
// ---- Bob vs Alice's resources: everything must be a 404 (or absent). ----
|
||||||
|
let bob_hits: &[(&str, String, Option<serde_json::Value>)] = &[
|
||||||
|
("GET", format!("/api/albums/{album_id}"), None),
|
||||||
|
(
|
||||||
|
"PATCH",
|
||||||
|
format!("/api/albums/{album_id}"),
|
||||||
|
Some(serde_json::json!({ "name": "stolen" })),
|
||||||
|
),
|
||||||
|
("DELETE", format!("/api/albums/{album_id}"), None),
|
||||||
|
("POST", format!("/api/albums/{album_id}/photos?filename=x.jpg"), None),
|
||||||
|
("GET", format!("/api/albums/{album_id}/shares"), None),
|
||||||
|
(
|
||||||
|
"POST",
|
||||||
|
format!("/api/albums/{album_id}/shares"),
|
||||||
|
Some(serde_json::json!({ "label": "x" })),
|
||||||
|
),
|
||||||
|
("POST", format!("/api/albums/{album_id}/zip"), None),
|
||||||
|
("GET", format!("/api/albums/{album_id}/photos/by-hash/hash-a"), None),
|
||||||
|
("DELETE", format!("/api/photos/{photo_id}"), None),
|
||||||
|
("POST", format!("/api/photos/{photo_id}/reprocess"), None),
|
||||||
|
("DELETE", format!("/api/shares/{share_id}"), None),
|
||||||
|
("POST", format!("/api/shares/{share_id}/reset-lock"), None),
|
||||||
|
];
|
||||||
|
for (method, path, body) in bob_hits {
|
||||||
|
let (status, _) = request(&router, method, path, Some(&cookie_b), body.clone()).await;
|
||||||
|
assert_eq!(
|
||||||
|
status,
|
||||||
|
StatusCode::NOT_FOUND,
|
||||||
|
"cross-tenant {method} {path} must 404"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
let (status, list) = request(&router, "GET", "/api/albums", Some(&cookie_b), None).await;
|
||||||
|
assert_eq!(status, StatusCode::OK);
|
||||||
|
assert_eq!(list.as_array().unwrap().len(), 0, "bob must see no albums");
|
||||||
|
|
||||||
|
// ---- Dual-auth image routes must ALSO reject a non-owning photographer.
|
||||||
|
// Bob owns no share and doesn't own the album, so authorize_photo falls
|
||||||
|
// through to the share-cookie path and 401s BEFORE any S3 access — the
|
||||||
|
// cross-tenant original leak these routes previously allowed. ----
|
||||||
|
for path in [
|
||||||
|
format!("/api/img/{photo_id}/thumb"),
|
||||||
|
format!("/api/img/{photo_id}/preview"),
|
||||||
|
format!("/api/photos/{photo_id}/original"),
|
||||||
|
] {
|
||||||
|
let (status, _) = request(&router, "GET", &path, Some(&cookie_b), None).await;
|
||||||
|
assert_eq!(
|
||||||
|
status,
|
||||||
|
StatusCode::UNAUTHORIZED,
|
||||||
|
"cross-tenant image GET {path} must not authorize"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---- Alice keeps full access to her own resources. ----
|
||||||
|
let (status, list) = request(&router, "GET", "/api/albums", Some(&cookie_a), None).await;
|
||||||
|
assert_eq!(status, StatusCode::OK);
|
||||||
|
assert_eq!(list.as_array().unwrap().len(), 1);
|
||||||
|
let (status, _) = request(
|
||||||
|
&router,
|
||||||
|
"GET",
|
||||||
|
&format!("/api/albums/{album_id}"),
|
||||||
|
Some(&cookie_a),
|
||||||
|
None,
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
assert_eq!(status, StatusCode::OK);
|
||||||
|
let (status, _) = request(
|
||||||
|
&router,
|
||||||
|
"GET",
|
||||||
|
&format!("/api/albums/{album_id}/photos/by-hash/hash-a"),
|
||||||
|
Some(&cookie_a),
|
||||||
|
None,
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
assert_eq!(status, StatusCode::OK);
|
||||||
|
// Ownership check runs before the body is read: empty upload = 400, not 404.
|
||||||
|
let (status, _) = request(
|
||||||
|
&router,
|
||||||
|
"POST",
|
||||||
|
&format!("/api/albums/{album_id}/photos?filename=x.jpg"),
|
||||||
|
Some(&cookie_a),
|
||||||
|
None,
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
assert_eq!(status, StatusCode::BAD_REQUEST);
|
||||||
|
// No ready photos yet: zip is a 400 for the owner, never an S3 call.
|
||||||
|
let (status, _) = request(
|
||||||
|
&router,
|
||||||
|
"POST",
|
||||||
|
&format!("/api/albums/{album_id}/zip"),
|
||||||
|
Some(&cookie_a),
|
||||||
|
None,
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
assert_eq!(status, StatusCode::BAD_REQUEST);
|
||||||
|
|
||||||
|
// ---- No session at all: 401 on the admin surface. ----
|
||||||
|
for path in ["/api/albums", "/api/me"] {
|
||||||
|
let (status, _) = request(&router, "GET", path, None, None).await;
|
||||||
|
assert_eq!(status, StatusCode::UNAUTHORIZED, "{path} without session");
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user