fix(api): resolve s3 copy source under endpoint path (#3422)

deploy/self-hosting/.env.example

@@ -157,8 +157,8 @@ MEILI_MASTER_KEY=CHANGE_ME
 #FLUXER_GEOIP_DB_PATH=
 
 # The client address. The edge sets X-Forwarded-For on every hop, so keep the
-# trust on and the default header. Turning the trust off makes the api refuse
-# every request outside its exempt routes with a 403.
+# trust on and the default header. Turning the trust off makes the api ignore
+# the header and use the connection's peer address, which here is the edge.
 #FLUXER_CLIENT_IP_HEADER_NAME=x-forwarded-for
 #FLUXER_TRUST_CLIENT_IP_HEADER=true
 

fluxer_api/pkgs/geoip/src/ResolveClientGeoip.ts

@@ -2,18 +2,11 @@
 
 import {AGE_BLOCKED_GEOS, AGE_RESTRICTED_GEOS} from '@fluxer/instance_bootstrap/src/AgeGeos';
 import type {GeolocationResponse} from '@fluxer/instance_bootstrap/src/Types';
-import {extractClientIp} from '@fluxer/ip_utils/src/ClientIp';
 import {lookupGeoipByIp} from '@pkgs/geoip/src/GeoipLookup';
 
-interface ResolveClientGeoipOptions {
-	maxmindDbPath: string | undefined;
-	trustClientIpHeader: boolean;
-	clientIpHeaderName: string;
-}
-
 export async function resolveClientGeoip(
-	req: Request,
-	options: ResolveClientGeoipOptions,
+	ip: string | null,
+	maxmindDbPath: string | undefined,
 ): Promise<GeolocationResponse> {
 	const response: GeolocationResponse = {
 		countryCode: null,
@@ -23,14 +16,10 @@ export async function resolveClientGeoip(
 		ageRestrictedGeos: AGE_RESTRICTED_GEOS,
 		ageBlockedGeos: AGE_BLOCKED_GEOS,
 	};
-	const ip = extractClientIp(req, {
-		trustClientIpHeader: options.trustClientIpHeader,
-		clientIpHeaderName: options.clientIpHeaderName,
-	});
 	if (!ip) {
 		return response;
 	}
-	const geoip = await lookupGeoipByIp(ip, options.maxmindDbPath);
+	const geoip = await lookupGeoipByIp(ip, maxmindDbPath);
 	response.countryCode = geoip.countryCode;
 	response.regionCode = geoip.regionCode ?? null;
 	response.latitude = geoip.latitude != null ? geoip.latitude.toString() : null;

fluxer_api/src/api/app/MiddlewarePipeline.ts

@@ -15,6 +15,7 @@ import {ServiceMiddleware} from '@app/api/middleware/ServiceMiddleware';
 import {TrustedClientIpHeaderMiddleware} from '@app/api/middleware/TrustedClientIpHeaderMiddleware';
 import {UserMiddleware} from '@app/api/middleware/UserMiddleware';
 import type {HonoApp} from '@app/api/types/HonoEnv';
+import {PeerAddressMiddleware} from '@app/api/utils/RequestClientIp';
 import {Headers as HttpHeaders} from '@fluxer/constants/src/Headers';
 import {InvalidApiOriginError} from '@fluxer/errors/src/domains/core/InvalidApiOriginError';
 import {cors} from '@fluxer/hono/src/middleware/Cors';
@@ -35,6 +36,7 @@ interface MiddlewarePipelineOptions {
 export function configureMiddleware(routes: HonoApp, options: MiddlewarePipelineOptions): void {
 	const {logger, nodeEnv, corsOrigins, trustClientIpHeader, clientIpHeaderName, maxInflightRequests} = options;
 	const resolvedHeader = resolveClientIpHeaderName(clientIpHeaderName);
+	routes.use(PeerAddressMiddleware);
 	routes.use('/webhooks/:webhook_id/:token', cors({origins: '*'}));
 	routes.use('/webhooks/:webhook_id/:token/messages/:message_id', cors({origins: '*'}));
 	routes.use(

fluxer_api/src/api/app/tests/ProductionClientIpPipeline.test.ts

@@ -2,31 +2,34 @@
 
 import {configureMiddleware} from '@app/api/app/MiddlewarePipeline';
 import {Config} from '@app/api/Config';
+import {InstanceController} from '@app/api/instance/InstanceController';
 import {setInjectedWorkerService} from '@app/api/middleware/ServiceRegistry';
 import {NoopLogger} from '@app/api/test/mocks/NoopLogger';
 import {NoopWorkerService} from '@app/api/test/NoopWorkerService';
 import type {HonoEnv} from '@app/api/types/HonoEnv';
+import {getRequestClientIp} from '@app/api/utils/RequestClientIp';
 import {AppErrorHandler, AppNotFoundHandler} from '@fluxer/errors/src/domains/core/ErrorHandlers';
 import {Hono} from 'hono';
 import {afterEach, beforeAll, beforeEach, describe, expect, it} from 'vitest';
 
 const CLIENT_IP_HEADER_NAME = 'x-real-ip';
 
-function createProductionApp(): Hono<HonoEnv> {
+function createProductionApp(trustClientIpHeader = true): Hono<HonoEnv> {
 	const routes = new Hono<HonoEnv>({strict: true});
 	configureMiddleware(routes, {
 		logger: new NoopLogger(),
 		nodeEnv: 'production',
 		corsOrigins: ['https://web.fluxer.app'],
-		trustClientIpHeader: true,
+		trustClientIpHeader,
 		clientIpHeaderName: CLIENT_IP_HEADER_NAME,
 		maxInflightRequests: 100,
 	});
 	routes.onError(AppErrorHandler);
 	routes.notFound(AppNotFoundHandler);
+	InstanceController(routes);
 	routes.post('/internal/rpc', (ctx) => ctx.json({ok: true}));
 	routes.get('/connections/bluesky/jwks.json', (ctx) => ctx.json({keys: []}));
-	routes.get('/users/@me', (ctx) => ctx.json({ok: true}));
+	routes.get('/users/@me', (ctx) => ctx.json({ok: true, ip: getRequestClientIp(ctx)}));
 	const app = new Hono<HonoEnv>({strict: true});
 	app.route('/v1', routes);
 	app.route('/', routes);
@@ -91,4 +94,52 @@ describe('client ip requirements across the production middleware pipeline', ()
 		expect(response.status).toBe(403);
 		expect(await response.json()).toMatchObject({code: 'FORBIDDEN'});
 	});
+
+	it('resolves the custom client ip header while it is trusted', async () => {
+		const app = createProductionApp();
+		const response = await app.request(
+			'http://api:8080/v1/users/@me',
+			{headers: {[CLIENT_IP_HEADER_NAME]: '203.0.113.10', 'x-forwarded-for': '198.51.100.1'}},
+			{incoming: {socket: {remoteAddress: '10.0.0.5'}}},
+		);
+		expect(response.status).toBe(200);
+		expect(await response.json()).toMatchObject({ip: '203.0.113.10'});
+	});
+
+	describe('with the client ip header untrusted', () => {
+		beforeEach(() => {
+			Config.proxy.trust_client_ip_header = false;
+		});
+
+		it('keys the request on the peer address and ignores a spoofed header', async () => {
+			const app = createProductionApp(false);
+			const response = await app.request(
+				'http://api:8080/v1/users/@me',
+				{headers: {[CLIENT_IP_HEADER_NAME]: '203.0.113.10', 'x-forwarded-for': '203.0.113.11'}},
+				{incoming: {socket: {remoteAddress: '::ffff:198.51.100.20'}}},
+			);
+			expect(response.status).toBe(200);
+			expect(await response.json()).toMatchObject({ip: '198.51.100.20'});
+		});
+
+		it('serves the discovery document', async () => {
+			const app = createProductionApp(false);
+			const response = await app.request(
+				'http://api:8080/.well-known/fluxer',
+				{headers: {'x-forwarded-for': '203.0.113.10'}},
+				{incoming: {socket: {remoteAddress: '198.51.100.20'}}},
+			);
+			expect(response.status).toBe(200);
+		});
+
+		it('serves the internal rpc route', async () => {
+			const app = createProductionApp(false);
+			const response = await app.request('http://api:8080/internal/rpc', {
+				method: 'POST',
+				headers: {'content-type': 'application/json'},
+				body: '{}',
+			});
+			expect(response.status).toBe(200);
+		});
+	});
 });

fluxer_api/src/api/geolocation/GeolocationController.ts

@@ -5,6 +5,7 @@ import {RateLimitMiddleware} from '@app/api/middleware/RateLimitMiddleware';
 import {OpenAPI} from '@app/api/middleware/ResponseTypeMiddleware';
 import {RateLimitConfigs} from '@app/api/RateLimitConfig';
 import type {HonoEnv} from '@app/api/types/HonoEnv';
+import {getRequestClientIp} from '@app/api/utils/RequestClientIp';
 import {GeolocationResponse} from '@fluxer/schema/src/domains/geolocation/GeolocationSchemas';
 import {resolveClientGeoip} from '@pkgs/geoip/src/ResolveClientGeoip';
 import type {Hono} from 'hono';
@@ -27,11 +28,7 @@ export function GeolocationController(app: Hono<HonoEnv>): void {
 			ctx.header('Access-Control-Allow-Origin', '*');
 			ctx.header('Access-Control-Allow-Headers', 'Content-Type');
 			ctx.header('Access-Control-Allow-Methods', 'GET, OPTIONS');
-			const response = await resolveClientGeoip(ctx.req.raw, {
-				maxmindDbPath: Config.geoip.maxmindDbPath,
-				trustClientIpHeader: Config.proxy.trust_client_ip_header,
-				clientIpHeaderName: Config.proxy.client_ip_header,
-			});
+			const response = await resolveClientGeoip(getRequestClientIp(ctx), Config.geoip.maxmindDbPath);
 			return ctx.json(response);
 		},
 	);

fluxer_api/src/api/infrastructure/StorageService.test.ts

@@ -225,3 +225,55 @@ describe('StorageService.listObjects', () => {
 		);
 	});
 });
+
+function copySourceFor(endpoint: string, forcePathStyle: boolean): Promise<unknown> {
+	const service = new StorageService({
+		endpoint,
+		forcePathStyle,
+		region: 'us-east-1',
+		accessKeyId: 'fluxer',
+		secretAccessKey: 'fluxer-secret',
+	});
+	let copySource: unknown;
+	Object.assign(service, {
+		client: {
+			send: async (command: {input: {CopySource?: unknown}}) => {
+				copySource = command.input.CopySource;
+				return {};
+			},
+		},
+	});
+	return service
+		.copyObject({
+			sourceBucket: 'fluxer-uploads',
+			sourceKey: 'a b/c+d.pdf',
+			destinationBucket: 'fluxer-cdn',
+			destinationKey: 'attachments/1/2/c.pdf',
+		})
+		.then(() => copySource);
+}
+
+describe('StorageService.copyObject copy source', () => {
+	it.each([
+		['http://seaweedfs:8333', true],
+		['http://seaweedfs:8333/', true],
+		['https://s3.example.test', false],
+	])(
+		'names only the bucket and key for an endpoint without a path (%s, path style %s)',
+		async (endpoint, pathStyle) => {
+			await expect(copySourceFor(endpoint, pathStyle)).resolves.toBe('fluxer-uploads/a%20b/c%2Bd.pdf');
+		},
+	);
+
+	it('puts the endpoint path in front of the bucket with path-style addressing', async () => {
+		await expect(copySourceFor('https://s3.example.test/real%20bucket/sub/', true)).resolves.toBe(
+			'real%20bucket/sub/fluxer-uploads/a%20b/c%2Bd.pdf',
+		);
+	});
+
+	it('puts the endpoint path between the bucket and the key with virtual-hosted addressing', async () => {
+		await expect(copySourceFor('https://s3.example.test/real%20bucket/sub', false)).resolves.toBe(
+			'fluxer-uploads/real%20bucket/sub/a%20b/c%2Bd.pdf',
+		);
+	});
+});

fluxer_api/src/api/infrastructure/StorageService.ts

@@ -115,6 +115,13 @@ const REJECTED_SERVER_SIDE_COPY_ERRORS = new Set([
 	'MethodNotAllowed',
 ]);
 
+function encodedEndpointPath(endpoint: string): Array<string> {
+	return (URL.parse(endpoint)?.pathname ?? '')
+		.split('/')
+		.filter((segment) => segment !== '')
+		.map((segment) => encodeURIComponent(decodeURIComponent(segment)));
+}
+
 function isRejectedServerSideCopy(error: unknown): boolean {
 	return (
 		error instanceof S3ServiceException &&
@@ -126,6 +133,7 @@ export class StorageService implements IStorageService {
 	private readonly client: S3Client;
 	private readonly presignClient: S3Client;
 	private readonly provider: S3ProviderSettings;
+	private readonly endpointPath: Array<string>;
 
 	constructor(provider?: S3ProviderSettings) {
 		this.provider = provider ?? {
@@ -136,6 +144,7 @@ export class StorageService implements IStorageService {
 			accessKeyId: Config.s3.accessKeyId,
 			secretAccessKey: Config.s3.secretAccessKey,
 		};
+		this.endpointPath = encodedEndpointPath(this.provider.endpoint);
 		this.client = buildPooledS3Client({
 			endpoint: this.provider.endpoint,
 			region: this.provider.region,
@@ -492,7 +501,7 @@ export class StorageService implements IStorageService {
 				new CopyObjectCommand({
 					Bucket: destinationBucket,
 					Key: destinationKey,
-					CopySource: `${encodeURIComponent(sourceBucket)}/${sourceKey.split('/').map(encodeURIComponent).join('/')}`,
+					CopySource: this.copySource(sourceBucket, sourceKey),
 					ContentType: newContentType,
 					MetadataDirective: newContentType ? 'REPLACE' : undefined,
 				}),
@@ -508,6 +517,14 @@ export class StorageService implements IStorageService {
 		}
 	}
 
+	private copySource(bucket: string, key: string): string {
+		const encodedBucket = encodeURIComponent(bucket);
+		const location = this.provider.forcePathStyle
+			? [...this.endpointPath, encodedBucket]
+			: [encodedBucket, ...this.endpointPath];
+		return [...location, ...key.split('/').map(encodeURIComponent)].join('/');
+	}
+
 	private async copyObjectThroughApi(
 		{
 			sourceBucket,

fluxer_api/src/api/infrastructure/StorageServiceEndpointPath.test.ts

@@ -0,0 +1,129 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import {StorageService} from '@app/api/infrastructure/StorageService';
+import {server} from '@app/api/test/msw/server';
+import {HttpResponse, http} from 'msw';
+import {beforeEach, describe, expect, it} from 'vitest';
+
+const HOST = 's3.single-bucket.test';
+const REAL_BUCKET = 'realbucket';
+
+function errorResponse(code: string, status: number): HttpResponse<string> {
+	return new HttpResponse(
+		`<?xml version="1.0" encoding="UTF-8"?><Error><Code>${code}</Code><Message></Message><RequestId>tx0</RequestId></Error>`,
+		{status, headers: {'Content-Type': 'application/xml'}},
+	);
+}
+
+function splitBucket(path: string): {bucket: string; key: string} {
+	const [bucket = '', ...key] = path.split('/');
+	return {bucket, key: key.join('/')};
+}
+
+function singleBucketStore() {
+	const objects = new Map<string, {body: Uint8Array; contentType: string}>();
+	const resolve = (url: URL): {bucket: string; key: string} => {
+		const path = decodeURIComponent(url.pathname.slice(1));
+		if (url.hostname === HOST) {
+			return splitBucket(path);
+		}
+		return {bucket: url.hostname.slice(0, -HOST.length - 1), key: path};
+	};
+	server.use(
+		http.all(/^https:\/\/([a-z0-9-]+\.)?s3\.single-bucket\.test\//u, async ({request}) => {
+			const target = resolve(new URL(request.url));
+			if (target.bucket !== REAL_BUCKET) {
+				return errorResponse('NoSuchBucket', 404);
+			}
+			if (request.method === 'GET') {
+				const object = objects.get(target.key);
+				if (!object) {
+					return errorResponse('NoSuchKey', 404);
+				}
+				return new HttpResponse(object.body, {
+					status: 200,
+					headers: {'Content-Type': object.contentType, 'Content-Length': String(object.body.length)},
+				});
+			}
+			if (request.method !== 'PUT') {
+				return errorResponse('NotImplemented', 501);
+			}
+			const copySource = request.headers.get('x-amz-copy-source');
+			if (!copySource) {
+				objects.set(target.key, {
+					body: new Uint8Array(await request.arrayBuffer()),
+					contentType: request.headers.get('content-type') ?? 'application/octet-stream',
+				});
+				return new HttpResponse(null, {status: 200, headers: {ETag: '"x"'}});
+			}
+			const source = splitBucket(decodeURIComponent(copySource.replace(/^\//u, '')));
+			if (source.bucket !== REAL_BUCKET) {
+				return errorResponse('NoSuchBucket', 404);
+			}
+			const object = objects.get(source.key);
+			if (!object) {
+				return errorResponse('NoSuchKey', 404);
+			}
+			objects.set(target.key, {...object, contentType: request.headers.get('content-type') ?? object.contentType});
+			return new HttpResponse(
+				'<?xml version="1.0" encoding="UTF-8"?><CopyObjectResult><ETag>"x"</ETag></CopyObjectResult>',
+				{status: 200, headers: {'Content-Type': 'application/xml'}},
+			);
+		}),
+	);
+	return objects;
+}
+
+function storage(endpoint: string, forcePathStyle: boolean): StorageService {
+	return new StorageService({
+		endpoint,
+		forcePathStyle,
+		region: 'us-east-1',
+		accessKeyId: 'TEST',
+		secretAccessKey: 'TEST',
+	});
+}
+
+const PDF = new TextEncoder().encode('%PDF-1.7\n1 0 obj << /Type /Catalog >> endobj\n%%EOF\n');
+
+describe('StorageService with every bucket inside one real bucket', () => {
+	let objects: ReturnType<typeof singleBucketStore>;
+
+	beforeEach(() => {
+		objects = singleBucketStore();
+	});
+
+	it('copies a non-media attachment when the endpoint names the bucket with path-style addressing', async () => {
+		const service = storage(`https://${HOST}/${REAL_BUCKET}`, true);
+		await service.uploadObject({bucket: 'fluxer-uploads', key: 'upload 1', body: PDF});
+
+		await expect(
+			service.copyObjectWithMetadataStripping({
+				sourceBucket: 'fluxer-uploads',
+				sourceKey: 'upload 1',
+				destinationBucket: 'fluxer-cdn',
+				destinationKey: 'attachments/1/2/file.pdf',
+				contentType: 'application/pdf',
+			}),
+		).resolves.toBeNull();
+
+		expect(objects.get('fluxer-cdn/attachments/1/2/file.pdf')?.contentType).toBe('application/pdf');
+		await expect(service.readObject('fluxer-cdn', 'attachments/1/2/file.pdf')).resolves.toEqual(PDF);
+	});
+
+	it('copies within a bucket whose virtual host keeps the endpoint path as a key prefix', async () => {
+		const service = storage(`https://${HOST}/prefix`, false);
+		await service.uploadObject({bucket: REAL_BUCKET, key: 'upload-2', body: PDF});
+
+		await service.copyObject({
+			sourceBucket: REAL_BUCKET,
+			sourceKey: 'upload-2',
+			destinationBucket: REAL_BUCKET,
+			destinationKey: 'attachments/1/3/file.pdf',
+			newContentType: 'application/pdf',
+		});
+
+		expect(objects.get('prefix/attachments/1/3/file.pdf')?.contentType).toBe('application/pdf');
+		await expect(service.readObject(REAL_BUCKET, 'attachments/1/3/file.pdf')).resolves.toEqual(PDF);
+	});
+});

fluxer_api/src/api/middleware/tests/RequireClientIpMiddleware.test.ts

@@ -3,19 +3,20 @@
 import {Config} from '@app/api/Config';
 import {RequireClientIpMiddleware} from '@app/api/middleware/RequireClientIpMiddleware';
 import type {HonoEnv} from '@app/api/types/HonoEnv';
-import type {ClientIpResolution} from '@app/api/utils/RequestClientIp';
+import {type ClientIpResolution, PeerAddressMiddleware} from '@app/api/utils/RequestClientIp';
 import {AppErrorHandler} from '@fluxer/errors/src/domains/core/ErrorHandlers';
 import {Hono} from 'hono';
 import {afterEach, beforeEach, describe, expect, it} from 'vitest';
 
 interface Harness {
-	request: (headers: Record<string, string>) => Promise<Response>;
+	request: (headers: Record<string, string>, peerAddress?: string) => Promise<Response>;
 	resolutions: Array<ClientIpResolution | undefined>;
 }
 
 function createHarness(path = 'http://localhost/v1/messages'): Harness {
 	const resolutions: Array<ClientIpResolution | undefined> = [];
 	const app = new Hono<HonoEnv>();
+	app.use(PeerAddressMiddleware);
 	app.use(RequireClientIpMiddleware());
 	app.get('/v1/messages', (ctx) => {
 		resolutions.push(ctx.get('clientIpResolution'));
@@ -25,7 +26,8 @@ function createHarness(path = 'http://localhost/v1/messages'): Harness {
 	app.get('/internal/rpc', (ctx) => ctx.text('OK'));
 	app.onError(AppErrorHandler);
 	return {
-		request: async (headers) => app.request(path, {headers}),
+		request: async (headers, peerAddress) =>
+			app.request(path, {headers}, peerAddress ? {incoming: {socket: {remoteAddress: peerAddress}}} : undefined),
 		resolutions,
 	};
 }
@@ -53,7 +55,15 @@ describe('RequireClientIpMiddleware', () => {
 		expect(await response.json()).toMatchObject({code: 'FORBIDDEN'});
 	});
 
-	it('rejects a request when the header is not trusted with 403', async () => {
+	it('passes an untrusted request on its peer address and ignores the header', async () => {
+		Config.proxy.trust_client_ip_header = false;
+		const harness = createHarness();
+		const response = await harness.request({'x-forwarded-for': '203.0.113.10'}, '::ffff:198.51.100.20');
+		expect(response.status).toBe(200);
+		expect(harness.resolutions[0]?.ip).toBe('198.51.100.20');
+	});
+
+	it('rejects an untrusted request with no peer address with 403', async () => {
 		Config.proxy.trust_client_ip_header = false;
 		const harness = createHarness();
 		const response = await harness.request({'x-forwarded-for': '203.0.113.10'});
@@ -61,6 +71,12 @@ describe('RequireClientIpMiddleware', () => {
 		expect(await response.json()).toMatchObject({code: 'FORBIDDEN'});
 	});
 
+	it('ignores the peer address while the header is trusted', async () => {
+		const harness = createHarness();
+		const response = await harness.request({}, '198.51.100.20');
+		expect(response.status).toBe(403);
+	});
+
 	it('rejects a request with no client ip header with 403 FORBIDDEN', async () => {
 		const harness = createHarness();
 		const response = await harness.request({});

fluxer_api/src/api/utils/RequestClientIp.ts

@@ -9,7 +9,9 @@ import {
 	requireClientIp,
 	resolveClientIpHeaderName,
 } from '@fluxer/ip_utils/src/ClientIp';
+import type {HttpBindings} from '@hono/node-server';
 import type {Context} from 'hono';
+import {createMiddleware} from 'hono/factory';
 
 export interface ClientIpResolution {
 	trustClientIpHeader: boolean;
@@ -22,6 +24,20 @@ interface ClientIpResolutionOptions {
 	clientIpHeaderName: string;
 }
 
+const peerAddresses = new WeakMap<Request, string>();
+
+export const PeerAddressMiddleware = createMiddleware<HonoEnv>(async (ctx, next) => {
+	const remoteAddress = (ctx.env as Partial<HttpBindings> | undefined)?.incoming?.socket?.remoteAddress;
+	if (remoteAddress) {
+		peerAddresses.set(ctx.req.raw, remoteAddress);
+	}
+	await next();
+});
+
+function withPeerAddress(request: Request, options: ClientIpResolutionOptions) {
+	return {...options, peerAddress: peerAddresses.get(request) ?? null};
+}
+
 export function resolveClientIpWithOptions(ctx: Context<HonoEnv>, options: ClientIpResolutionOptions): string | null {
 	const clientIpHeaderName = resolveClientIpHeaderName(options.clientIpHeaderName);
 	const {trustClientIpHeader} = options;
@@ -33,7 +49,7 @@ export function resolveClientIpWithOptions(ctx: Context<HonoEnv>, options: Clien
 	) {
 		return cached.ip;
 	}
-	const ip = extractClientIp(ctx.req.raw, {trustClientIpHeader, clientIpHeaderName});
+	const ip = extractClientIp(ctx.req.raw, withPeerAddress(ctx.req.raw, {trustClientIpHeader, clientIpHeaderName}));
 	ctx.set('clientIpResolution', {trustClientIpHeader, clientIpHeaderName, ip});
 	return ip;
 }
@@ -43,11 +59,11 @@ function proxyClientIpOptions(proxy: APIConfig['proxy']): ClientIpResolutionOpti
 }
 
 export function extractConfiguredClientIp(request: Request): string | null {
-	return extractClientIp(request, proxyClientIpOptions(Config.proxy));
+	return extractClientIp(request, withPeerAddress(request, proxyClientIpOptions(Config.proxy)));
 }
 
 export function requireConfiguredClientIp(request: Request, proxy: APIConfig['proxy']): string {
-	return requireClientIp(request, proxyClientIpOptions(proxy));
+	return requireClientIp(request, withPeerAddress(request, proxyClientIpOptions(proxy)));
 }
 
 export function getRequestClientIp(ctx: Context<HonoEnv>): string | null {

fluxer_api/src/api/utils/tests/RequestClientIp.test.ts

@@ -1,11 +1,19 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
+import {Config} from '@app/api/Config';
 import type {HonoEnv} from '@app/api/types/HonoEnv';
-import {getRequestClientIp, requireRequestClientIp, resolveClientIpWithOptions} from '@app/api/utils/RequestClientIp';
+import {
+	extractConfiguredClientIp,
+	getRequestClientIp,
+	PeerAddressMiddleware,
+	requireConfiguredClientIp,
+	requireRequestClientIp,
+	resolveClientIpWithOptions,
+} from '@app/api/utils/RequestClientIp';
 import {AppErrorHandler} from '@fluxer/errors/src/domains/core/ErrorHandlers';
 import {MissingClientIpError} from '@fluxer/ip_utils/src/ClientIp';
 import {Hono} from 'hono';
-import {describe, expect, it} from 'vitest';
+import {afterEach, describe, expect, it} from 'vitest';
 
 interface CountingRequest {
 	request: Request;
@@ -28,6 +36,12 @@ function createCountingRequest(headers: Record<string, string>): CountingRequest
 }
 
 describe('RequestClientIp', () => {
+	const previousTrustClientIpHeader = Config.proxy.trust_client_ip_header;
+
+	afterEach(() => {
+		Config.proxy.trust_client_ip_header = previousTrustClientIpHeader;
+	});
+
 	it('normalizes the resolved address the same way for every consumer', async () => {
 		const seen: Array<string | null> = [];
 		const app = new Hono<HonoEnv>();
@@ -102,4 +116,22 @@ describe('RequestClientIp', () => {
 		expect(response.status).toBe(403);
 		expect(await response.json()).toMatchObject({code: 'FORBIDDEN'});
 	});
+	it('resolves the peer address for request only callers when the header is untrusted', async () => {
+		Config.proxy.trust_client_ip_header = false;
+		const seen: Array<string | null> = [];
+		const app = new Hono<HonoEnv>();
+		app.use(PeerAddressMiddleware);
+		app.get('/v1/messages', (ctx) => {
+			seen.push(getRequestClientIp(ctx));
+			seen.push(extractConfiguredClientIp(ctx.req.raw));
+			seen.push(requireConfiguredClientIp(ctx.req.raw, Config.proxy));
+			return ctx.text('ok');
+		});
+		await app.request(
+			'http://localhost/v1/messages',
+			{headers: {'x-forwarded-for': '203.0.113.10'}},
+			{incoming: {socket: {remoteAddress: '::ffff:198.51.100.20'}}},
+		);
+		expect(seen).toEqual(['198.51.100.20', '198.51.100.20', '198.51.100.20']);
+	});
 });

fluxer_desktop/src/main/DesktopOutboundHTTP.test.mjs

@@ -65,7 +65,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => {
 				headers: null,
 				method: 'GET',
 				originTrust: DesktopOriginTrust.REGISTERED,
-				serviceName: 'test',
 				signal: null,
 				timeoutMs: 500,
 				url: 'https://media.internal/avatars/1.webp',
@@ -104,7 +103,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => {
 				headers: null,
 				method: 'GET',
 				originTrust: DesktopOriginTrust.REGISTERED,
-				serviceName: 'test',
 				signal: null,
 				timeoutMs: 500,
 				url: 'https://media.internal/avatars/1.webp',
@@ -161,7 +159,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => {
 				headers: null,
 				method: 'GET',
 				originTrust: DesktopOriginTrust.REGISTERED,
-				serviceName: 'test',
 				signal: null,
 				timeoutMs: 500,
 				url: 'https://media.mixed.example/avatars/1.webp',
@@ -193,44 +190,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => {
 		outboundHTTP.cleanup();
 	});
 
-	test('a request that cannot be constructed releases its in-flight slot', async () => {
-		const {port, server} = await startServer({
-			'/ok': (_request, response) => sendImage(response, 'image/png', PNG_BYTES),
-		});
-		try {
-			const outboundHTTP = new DesktopOutboundHTTP({
-				resolveHostAddresses: stubResolver({'lan.example': ['127.0.0.1']}),
-			});
-			await outboundHTTP.registerAnchoredOrigins({
-				anchorOrigin: `http://lan.example:${port}`,
-				origins: [`http://lan.example:${port}`],
-			});
-			const send = (headers) =>
-				outboundHTTP.request({
-					body: null,
-					expectedOrigin: `http://lan.example:${port}`,
-					headers,
-					method: 'GET',
-					originTrust: DesktopOriginTrust.REGISTERED,
-					serviceName: 'slot-leak',
-					signal: null,
-					timeoutMs: 2000,
-					url: `http://lan.example:${port}/ok`,
-				});
-
-			for (let attempt = 0; attempt < 80; attempt += 1) {
-				await assert.rejects(send({'x-bad': 'a\u0001b'}));
-			}
-			const message = await send(null);
-
-			assert.equal(message.status, 200);
-			message.message.resume();
-			outboundHTTP.cleanup();
-		} finally {
-			server.close();
-		}
-	});
-
 	test('a request whose signal already aborted is refused instead of sent', async () => {
 		const seen = [];
 		const {port, server} = await startServer({
@@ -257,7 +216,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => {
 					headers: null,
 					method: 'GET',
 					originTrust: DesktopOriginTrust.REGISTERED,
-					serviceName: 'aborted',
 					signal: controller.signal,
 					timeoutMs: 2000,
 					url: `http://lan.example:${port}/ok`,
@@ -289,7 +247,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => {
 				headers: null,
 				method: 'GET',
 				originTrust: DesktopOriginTrust.REGISTERED,
-				serviceName: 'test',
 				signal: null,
 				timeoutMs: 2000,
 				url: `http://media.lan.example:${port}/media/ok`,
@@ -331,7 +288,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => {
 					headers: {'content-type': 'application/json', 'x-fluxer-sudo-mode-jwt': 'previous'},
 					method,
 					originTrust: DesktopOriginTrust.REGISTERED,
-					serviceName: 'test',
 					signal: null,
 					timeoutMs: 2000,
 					url: `http://lan.example:${port}/api/v1/oauth2/applications/1`,
@@ -549,7 +505,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => {
 			headers: null,
 			method: 'GET',
 			originTrust: DesktopOriginTrust.BOUND,
-			serviceName: 'test',
 			signal,
 			timeoutMs: 2000,
 			url: `https://home.example:${port}/ping`,
@@ -577,7 +532,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => {
 			headers: null,
 			method: 'GET',
 			originTrust: DesktopOriginTrust.BOUND,
-			serviceName: 'test',
 			signal: null,
 			timeoutMs: 200,
 			url: 'http://blackhole.example:9/ping',
@@ -606,7 +560,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => {
 				headers: null,
 				method: 'GET',
 				originTrust: DesktopOriginTrust.BOUND,
-				serviceName: 'test',
 				signal: null,
 				timeoutMs: 2000,
 				url: `http://mixed.example:${port}/ping`,
@@ -632,7 +585,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => {
 					headers: null,
 					method: 'GET',
 					originTrust: DesktopOriginTrust.BOUND,
-					serviceName: 'test',
 					signal: null,
 					timeoutMs: 2000,
 					url: `http://127.0.0.1:${other.port}/steal`,
@@ -670,7 +622,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => {
 					headers: null,
 					method: 'GET',
 					originTrust: DesktopOriginTrust.BOUND,
-					serviceName: 'test',
 					signal: null,
 					timeoutMs: 5000,
 					url: `http://127.0.0.1:${port}/ping`,
@@ -720,7 +671,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => {
 					headers: {'content-type': 'text/plain'},
 					method: 'PUT',
 					originTrust: DesktopOriginTrust.BOUND,
-					serviceName: 'test',
 					signal: null,
 					timeoutMs: 5000,
 					url: `http://127.0.0.1:${port}/upload`,
@@ -775,7 +725,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => {
 					headers: null,
 					method: 'GET',
 					originTrust: DesktopOriginTrust.REGISTERED,
-					serviceName: 'test',
 					signal: null,
 					timeoutMs: 2000,
 					url: 'https://unrouted.example/',
@@ -826,7 +775,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => {
 				headers: null,
 				method: 'GET',
 				originTrust: DesktopOriginTrust.REGISTERED,
-				serviceName: 'test',
 				signal: null,
 				timeoutMs: 2000,
 				url: `http://dual.example:${port}/ping`,
@@ -857,7 +805,6 @@ describe('DesktopOutboundHTTP origin anchoring', () => {
 				headers: null,
 				method: 'GET',
 				originTrust: DesktopOriginTrust.REGISTERED,
-				serviceName: 'test',
 				signal: null,
 				timeoutMs: 3000,
 				url: `http://multi.example:${port}/ping`,
@@ -1007,7 +954,6 @@ describe('DesktopOutboundHTTP system proxy routes', () => {
 			headers: {Accept: 'image/png'},
 			method: 'GET',
 			originTrust: DesktopOriginTrust.REGISTERED,
-			serviceName: 'test',
 			signal: null,
 			timeoutMs: 500,
 			url: 'https://media.blocked.example/avatar.png',
@@ -1061,7 +1007,6 @@ describe('DesktopOutboundHTTP system proxy routes', () => {
 			headers: null,
 			method: 'GET',
 			originTrust,
-			serviceName: 'test',
 			signal: null,
 			timeoutMs: 500,
 			url,

fluxer_desktop/src/main/DesktopOutboundHTTP.ts

@@ -28,8 +28,6 @@ const DESKTOP_OUTBOUND_HTTP_RESOLUTION_TIMEOUT_MS = 10_000;
 const DESKTOP_OUTBOUND_HTTP_MAX_TARGET_URL_BYTES = 16 * 1024;
 const DESKTOP_OUTBOUND_HTTP_MAX_REDIRECT_LOCATION_BYTES = 4096;
 const DESKTOP_OUTBOUND_HTTP_MAX_ORIGINS = 2048;
-const DESKTOP_OUTBOUND_HTTP_MAX_ORIGIN_REQUESTS_IN_FLIGHT = 128;
-const DESKTOP_OUTBOUND_HTTP_MAX_IN_FLIGHT_PER_SERVICE = 64;
 const DESKTOP_OUTBOUND_HTTP_MAX_PENDING_RESOLUTIONS = 64;
 const DESKTOP_OUTBOUND_HTTP_MAX_REQUEST_BODY_BYTES = 1024 * 1024;
 const DESKTOP_OUTBOUND_HTTP_STALE_SOCKET_ATTEMPTS = 1;
@@ -113,7 +111,6 @@ export interface DesktopOutboundHTTPRequest {
 	readonly maximumRequestBodyBytes?: number;
 	readonly method: string;
 	readonly originTrust: DesktopOriginTrust;
-	readonly serviceName: string;
 	readonly signal: AbortSignal | null;
 	readonly timeoutMs: number;
 	readonly url: string;
@@ -565,7 +562,7 @@ interface DesktopProxiedSend {
 	readonly body: Uint8Array | Readable | null;
 	readonly headers: Readonly<Record<string, string>>;
 	readonly method: string;
-	readonly release: () => void;
+	readonly release?: () => void;
 	readonly signal: AbortSignal | null;
 	readonly target: URL;
 	readonly timeout: () => Error;
@@ -578,20 +575,15 @@ export class DesktopOutboundHTTP {
 		maxSockets: DESKTOP_OUTBOUND_HTTP_MAX_SOCKETS,
 	});
 	private readonly httpsAgent: https.Agent;
-	private readonly originRequestHttpAgent = new HappyEyeballsHttpAgent({
-		keepAlive: true,
-		maxSockets: DESKTOP_OUTBOUND_HTTP_MAX_IN_FLIGHT_PER_SERVICE,
-	});
+	private readonly originRequestHttpAgent = new HappyEyeballsHttpAgent({keepAlive: true});
 	private readonly originRequestHttpsAgent: https.Agent;
 	private readonly bindings = new Map<string, DesktopOriginAddressBinding>();
 	private readonly pendingBindings = new Map<string, DesktopPendingOriginBinding>();
 	private readonly registeredRequirements = new Map<string, DesktopAddressRequirement>();
-	private readonly originRequestsInFlightByService = new Map<string, number>();
 	private readonly resolveHostAddresses: DesktopHostAddressResolver;
 	private readonly resolveProxy: DesktopProxyResolver;
 	private readonly sendThroughSession: DesktopSessionHTTPSender;
 	private inFlight = 0;
-	private originRequestsInFlight = 0;
 	private activeResolutions = 0;
 	private acceptingRequests = true;
 
@@ -608,7 +600,6 @@ export class DesktopOutboundHTTP {
 		});
 		this.originRequestHttpsAgent = new HappyEyeballsHttpsAgent({
 			keepAlive: true,
-			maxSockets: DESKTOP_OUTBOUND_HTTP_MAX_IN_FLIGHT_PER_SERVICE,
 			...trust,
 		});
 	}
@@ -783,7 +774,6 @@ export class DesktopOutboundHTTP {
 		this.originRequestHttpsAgent.destroy();
 		this.bindings.clear();
 		this.registeredRequirements.clear();
-		this.originRequestsInFlightByService.clear();
 	}
 
 	private async routesDirect(url: URL): Promise<boolean> {
@@ -813,15 +803,6 @@ export class DesktopOutboundHTTP {
 		request: DesktopOutboundHTTPRequest,
 		target: URL,
 	): Promise<DesktopOutboundHTTPMessage> {
-		this.acquireSlot(request.serviceName);
-		let released = false;
-		const release = (): void => {
-			if (released) {
-				return;
-			}
-			released = true;
-			this.releaseSlot(request.serviceName);
-		};
 		return this.sendProxied({
 			body: this.proxiedRequestBody(request),
 			headers: requestHeaders(request.headers),
@@ -830,7 +811,6 @@ export class DesktopOutboundHTTP {
 			target,
 			timeout: () => new DesktopOutboundHTTPRequestTimeoutError(target.toString(), request.timeoutMs),
 			timeoutMs: request.timeoutMs,
-			release,
 		}).catch((error: unknown) => {
 			if (request.signal?.aborted === true) {
 				throw new DesktopOutboundHTTPRequestAbortedError(target.toString());
@@ -854,7 +834,7 @@ export class DesktopOutboundHTTP {
 		timer.unref();
 		const settle = (): void => {
 			clearTimeout(timer);
-			release();
+			release?.();
 		};
 		try {
 			const response = await this.sendThroughSession({
@@ -1083,42 +1063,11 @@ export class DesktopOutboundHTTP {
 		}
 	}
 
-	private acquireSlot(serviceName: string): void {
-		if (this.originRequestsInFlight >= DESKTOP_OUTBOUND_HTTP_MAX_ORIGIN_REQUESTS_IN_FLIGHT) {
-			throw new DesktopOutboundHTTPCapacityError('in-flight request');
-		}
-		const perService = this.originRequestsInFlightByService.get(serviceName) ?? 0;
-		if (perService >= DESKTOP_OUTBOUND_HTTP_MAX_IN_FLIGHT_PER_SERVICE) {
-			throw new DesktopOutboundHTTPCapacityError(`in-flight request for ${serviceName}`);
-		}
-		this.originRequestsInFlight += 1;
-		this.originRequestsInFlightByService.set(serviceName, perService + 1);
-	}
-
-	private releaseSlot(serviceName: string): void {
-		this.originRequestsInFlight -= 1;
-		const perService = (this.originRequestsInFlightByService.get(serviceName) ?? 1) - 1;
-		if (perService <= 0) {
-			this.originRequestsInFlightByService.delete(serviceName);
-		} else {
-			this.originRequestsInFlightByService.set(serviceName, perService);
-		}
-	}
-
 	private issueOriginRequest(
 		request: DesktopOutboundHTTPRequest,
 		target: URL,
 		binding: DesktopOriginAddressBinding,
 	): Promise<DesktopOutboundHTTPMessage> {
-		this.acquireSlot(request.serviceName);
-		let released = false;
-		const release = (): void => {
-			if (released) {
-				return;
-			}
-			released = true;
-			this.releaseSlot(request.serviceName);
-		};
 		return new Promise<DesktopOutboundHTTPMessage>((resolve, reject) => {
 			let attemptsRemaining = isReplayableRequest(request) ? DESKTOP_OUTBOUND_HTTP_STALE_SOCKET_ATTEMPTS : 0;
 			let active: http.ClientRequest | null = null;
@@ -1144,7 +1093,6 @@ export class DesktopOutboundHTTP {
 			const settle = (): void => {
 				clearTimeout(timeout);
 				request.signal?.removeEventListener('abort', onAbort);
-				release();
 			};
 			const destroyActive = (error: Error): void => {
 				active?.destroy(error);

fluxer_desktop/src/main/DesktopRuntimeSecurity.test.mjs

@@ -16,10 +16,12 @@ function createLogger() {
 
 function createSession() {
 	const session = {
+		requestFilter: null,
 		requestListener: null,
 		webRequest: {
-			onBeforeSendHeaders: (listener) => {
-				session.requestListener = listener;
+			onBeforeSendHeaders: (...args) => {
+				session.requestFilter = args.length > 1 ? args[0] : null;
+				session.requestListener = args.at(-1);
 			},
 		},
 	};
@@ -141,6 +143,12 @@ describe('DesktopRuntimeSecurity', () => {
 		);
 	});
 
+	test('only intercepts local app requests and the YouTube embed frame', () => {
+		const {session} = createInstalled();
+
+		assert.deepEqual(session.requestFilter, {urls: ['fluxer-app://*/*', 'https://www.youtube.com/embed/*']});
+	});
+
 	test('installs once per session and stops listening after cleanup', () => {
 		const {security, session} = createInstalled();
 		const firstRequestListener = session.requestListener;

fluxer_desktop/src/main/DesktopRuntimeSecurity.ts

@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
-import {STABLE_APP_URL} from '@electron/common/Constants';
+import {DESKTOP_APP_SCHEME, STABLE_APP_URL} from '@electron/common/Constants';
 import {isPlainRecord} from '@electron/common/PlainRecord';
 
 const SUB_FRAME_RESOURCE_TYPE = 'subFrame';
@@ -9,6 +9,10 @@ const REFERER_REQUEST_HEADER = 'Referer';
 const REFERER_HEADER_NAME_LOWERCASE = 'referer';
 const YOUTUBE_EMBED_FRAME_ORIGIN = 'https://www.youtube.com';
 const YOUTUBE_EMBED_FRAME_PATH_PREFIX = '/embed/';
+const DESKTOP_RUNTIME_SECURITY_REQUEST_URL_PATTERNS = [
+	`${DESKTOP_APP_SCHEME}://*/*`,
+	`${YOUTUBE_EMBED_FRAME_ORIGIN}${YOUTUBE_EMBED_FRAME_PATH_PREFIX}*`,
+];
 
 interface DesktopRuntimeSecurityLogger {
 	error: (...args: Array<unknown>) => void;
@@ -62,9 +66,12 @@ export class DesktopRuntimeSecurity {
 		}
 		this.installedSessions.add(session);
 		try {
-			session.webRequest.onBeforeSendHeaders((details: unknown, callback: unknown) => {
-				this.settleRequestHeaders(details, callback);
-			});
+			session.webRequest.onBeforeSendHeaders(
+				{urls: DESKTOP_RUNTIME_SECURITY_REQUEST_URL_PATTERNS},
+				(details: unknown, callback: unknown) => {
+					this.settleRequestHeaders(details, callback);
+				},
+			);
 		} catch (installationError) {
 			this.uninstall(session);
 			throw installationError;

fluxer_desktop/src/main/DesktopTrustedCertificates.test.mjs

@@ -41,7 +41,6 @@ async function fetchThrough(certificates, port) {
 			headers: null,
 			method: 'GET',
 			originTrust: DesktopOriginTrust.REGISTERED,
-			serviceName: 'test',
 			signal: null,
 			timeoutMs: 3000,
 			url: `https://ca.example:${port}/`,

fluxer_desktop/src/main/LocalAppProxyClient.ts

@@ -20,8 +20,6 @@ const LOCAL_APP_PROXY_REQUEST_BODY_MAX_BYTES = 100 * 1024 * 1024;
 export const LOCAL_APP_PROXY_ACCEPT_ENCODING = 'gzip, br';
 
 const LOCAL_APP_PROXY_RESOURCE_RESPONSE_HEADERS_DEADLINE_MS = 30_000;
-const LOCAL_APP_PROXY_MAX_CONCURRENT_REQUESTS = 32;
-const LOCAL_APP_PROXY_MAX_QUEUED_REQUESTS = 4096;
 const UPLOAD_PROGRESS_MIN_BYTES = 64 * 1024;
 const UPLOAD_PROGRESS_MIN_INTERVAL_MS = 100;
 const MINIMUM_RESPONSE_STATUS = 200;
@@ -42,101 +40,6 @@ class LocalAppProxyResponseHeadersDeadlineError extends Error {
 	}
 }
 
-class LocalAppProxyQueueOverflowError extends Error {
-	public constructor() {
-		super('Local app proxy has too many queued requests');
-		this.name = 'LocalAppProxyQueueOverflowError';
-	}
-}
-
-class LocalAppProxyConcurrencyPermit {
-	private releasePermit: (() => void) | null;
-
-	public constructor(releasePermit: () => void) {
-		this.releasePermit = releasePermit;
-	}
-
-	public release(): void {
-		const releasePermit = this.releasePermit;
-		if (releasePermit == null) {
-			return;
-		}
-		this.releasePermit = null;
-		releasePermit();
-	}
-}
-
-class LocalAppProxyResponseBodyLease {
-	public readonly reader: ReadableStreamDefaultReader<Uint8Array>;
-	private readonly permit: LocalAppProxyConcurrencyPermit;
-	private settled = false;
-
-	public constructor(body: ReadableStream<Uint8Array>, permit: LocalAppProxyConcurrencyPermit) {
-		this.reader = body.getReader();
-		this.permit = permit;
-	}
-
-	public release(): void {
-		if (this.settled) {
-			return;
-		}
-		this.settled = true;
-		try {
-			this.reader.releaseLock();
-		} finally {
-			this.permit.release();
-		}
-	}
-}
-
-class LocalAppProxyConcurrencyGate {
-	private active = 0;
-	private readonly waiting: Array<{readonly admit: () => void}> = [];
-	private readonly newestFirst: boolean;
-
-	public constructor(newestFirst: boolean) {
-		this.newestFirst = newestFirst;
-	}
-
-	public acquire(signal: AbortSignal): Promise<LocalAppProxyConcurrencyPermit> {
-		signal.throwIfAborted();
-		if (this.active < LOCAL_APP_PROXY_MAX_CONCURRENT_REQUESTS) {
-			this.active += 1;
-			return Promise.resolve(new LocalAppProxyConcurrencyPermit(() => this.release()));
-		}
-		if (this.waiting.length >= LOCAL_APP_PROXY_MAX_QUEUED_REQUESTS) {
-			throw new LocalAppProxyQueueOverflowError();
-		}
-		return new Promise<LocalAppProxyConcurrencyPermit>((resolve, reject) => {
-			let waiter: (typeof this.waiting)[number];
-			const onAbort = () => {
-				const index = this.waiting.indexOf(waiter);
-				if (index === -1) {
-					return;
-				}
-				this.waiting.splice(index, 1);
-				reject(signal.reason);
-			};
-			const admit = () => {
-				signal.removeEventListener('abort', onAbort);
-				resolve(new LocalAppProxyConcurrencyPermit(() => this.release()));
-			};
-			waiter = {admit};
-			this.waiting.push(waiter);
-			signal.addEventListener('abort', onAbort, {once: true});
-		});
-	}
-
-	private release(): void {
-		const next = this.newestFirst ? this.waiting.pop() : this.waiting.shift();
-		if (next != null) {
-			next.admit();
-			return;
-		}
-		this.active -= 1;
-	}
-}
-
 interface DesktopLocalAppProxyClientDependencies {
 	readonly outboundHTTP: DesktopOutboundHTTP;
 	readonly traffic: LocalAppProxyTraffic;
@@ -226,35 +129,20 @@ function emitLocalAppUploadProgress(progress: DesktopLocalAppUploadProgress): vo
 
 export class DesktopLocalAppProxyClient {
 	private readonly outboundHTTP: DesktopOutboundHTTP;
-	private readonly traffic: LocalAppProxyTraffic;
-	private readonly concurrency: LocalAppProxyConcurrencyGate;
 	private readonly responseHeadersDeadlineMs: number | null;
 
 	public constructor(dependencies: DesktopLocalAppProxyClientDependencies) {
 		this.outboundHTTP = dependencies.outboundHTTP;
-		this.traffic = dependencies.traffic;
-		const remoteResource = dependencies.traffic === LocalAppProxyTraffic.REMOTE_RESOURCE;
-		this.concurrency = new LocalAppProxyConcurrencyGate(remoteResource);
-		this.responseHeadersDeadlineMs = remoteResource ? LOCAL_APP_PROXY_RESOURCE_RESPONSE_HEADERS_DEADLINE_MS : null;
+		this.responseHeadersDeadlineMs =
+			dependencies.traffic === LocalAppProxyTraffic.REMOTE_RESOURCE
+				? LOCAL_APP_PROXY_RESOURCE_RESPONSE_HEADERS_DEADLINE_MS
+				: null;
 	}
 
 	public async fetch(request: LocalAppProxyFetchRequest): Promise<Response> {
-		const permit = await this.concurrency.acquire(request.signal);
-		try {
-			return await this.runFetchWithinHeadersDeadline(request, permit);
-		} catch (error) {
-			permit.release();
-			throw error;
-		}
-	}
-
-	private async runFetchWithinHeadersDeadline(
-		request: LocalAppProxyFetchRequest,
-		permit: LocalAppProxyConcurrencyPermit,
-	): Promise<Response> {
 		const deadlineMs = this.responseHeadersDeadlineMs;
 		if (deadlineMs == null || requestHasBody(request.method)) {
-			return await this.runFetch(request, permit);
+			return await this.runFetch(request);
 		}
 		const deadline = new AbortController();
 		const timer = setTimeout(
@@ -263,16 +151,13 @@ export class DesktopLocalAppProxyClient {
 		);
 		timer.unref();
 		try {
-			return await this.runFetch({...request, signal: AbortSignal.any([request.signal, deadline.signal])}, permit);
+			return await this.runFetch({...request, signal: AbortSignal.any([request.signal, deadline.signal])});
 		} finally {
 			clearTimeout(timer);
 		}
 	}
 
-	private async runFetch(
-		request: LocalAppProxyFetchRequest,
-		permit: LocalAppProxyConcurrencyPermit,
-	): Promise<Response> {
+	private async runFetch(request: LocalAppProxyFetchRequest): Promise<Response> {
 		const target = new URL(request.targetURL);
 		const body = requestHasBody(request.method) ? request.body : null;
 		const outboundRequest: DesktopOutboundHTTPRequest = {
@@ -282,12 +167,11 @@ export class DesktopLocalAppProxyClient {
 			maximumRequestBodyBytes: LOCAL_APP_PROXY_REQUEST_BODY_MAX_BYTES,
 			method: request.method,
 			originTrust: DesktopOriginTrust.REGISTERED,
-			serviceName: this.traffic,
 			signal: request.signal,
 			timeoutMs: LOCAL_APP_PROXY_DEADLINE_MS,
 			url: target.href,
 		};
-		return buildProxyResponse(await this.outboundHTTP.request(outboundRequest), request.method, permit);
+		return buildProxyResponse(await this.outboundHTTP.request(outboundRequest), request.method);
 	}
 
 	public async settleFailure({
@@ -368,11 +252,7 @@ function instrumentedRequestBody(
 	);
 }
 
-function buildProxyResponse(
-	message: DesktopOutboundHTTPMessage,
-	method: string,
-	permit: LocalAppProxyConcurrencyPermit,
-): Response {
+function buildProxyResponse(message: DesktopOutboundHTTPMessage, method: string): Response {
 	try {
 		const status = message.status;
 		if (status < MINIMUM_RESPONSE_STATUS || status > MAXIMUM_RESPONSE_STATUS) {
@@ -381,59 +261,20 @@ function buildProxyResponse(
 		const headers = responseHeaders(message.headers);
 		if (method === LOCAL_APP_PROXY_HEAD_METHOD || NULL_BODY_STATUSES.has(status)) {
 			message.message.resume();
-			permit.release();
 			return new Response(null, {status, statusText: message.statusText, headers});
 		}
 		const decoded = decodedMessageStream(message.message, contentEncoding(message.headers));
-		const body = holdPermitForResponseBody(Readable.toWeb(decoded) as ReadableStream<Uint8Array>, permit);
-		return new Response(body, {
+		return new Response(Readable.toWeb(decoded) as ReadableStream<Uint8Array>, {
 			status,
 			statusText: message.statusText,
 			headers,
 		});
 	} catch (error) {
 		message.message.destroy();
-		permit.release();
 		throw error;
 	}
 }
 
-function holdPermitForResponseBody(
-	body: ReadableStream<Uint8Array>,
-	permit: LocalAppProxyConcurrencyPermit,
-): ReadableStream<Uint8Array> {
-	const lease = new LocalAppProxyResponseBodyLease(body, permit);
-	return new ReadableStream<Uint8Array>({
-		async pull(controller) {
-			try {
-				const chunk = await lease.reader.read();
-				if (chunk.done) {
-					try {
-						controller.close();
-					} finally {
-						lease.release();
-					}
-					return;
-				}
-				controller.enqueue(chunk.value);
-			} catch (error) {
-				try {
-					controller.error(error);
-				} finally {
-					lease.release();
-				}
-			}
-		},
-		async cancel(reason) {
-			try {
-				await lease.reader.cancel(reason);
-			} finally {
-				lease.release();
-			}
-		},
-	});
-}
-
 function responseHeaders(incoming: http.IncomingHttpHeaders): Headers {
 	const headers = new Headers();
 	for (const [name, value] of Object.entries(incoming)) {

fluxer_desktop/src/main/LocalAppProxyConcurrency.test.mjs

@@ -2,7 +2,7 @@
 
 import assert from 'node:assert/strict';
 import {Buffer} from 'node:buffer';
-import {Readable} from 'node:stream';
+import {PassThrough, Readable} from 'node:stream';
 import {describe, mock, test} from 'node:test';
 import {installElectronStub} from './LocalAppTestSupport.test.mjs';
 
@@ -10,17 +10,9 @@ installElectronStub();
 
 const {DesktopLocalAppProxyClient, LocalAppProxyTraffic} = await import('./LocalAppProxyClient.ts');
 
-const PROXY_MAX_CONCURRENT_REQUESTS = 32;
-const PROXY_MAX_QUEUED_REQUESTS = 4096;
-const OUTBOUND_MAX_IN_FLIGHT_PER_SERVICE = 64;
 const RENDERED_AVATAR_COUNT = 120;
-
-class DesktopOutboundHTTPCapacityError extends Error {
-	constructor(limit) {
-		super(`Desktop outbound HTTP exceeded its ${limit} limit`);
-		this.name = 'DesktopOutboundHTTPCapacityError';
-	}
-}
+const PARKED_MEDIA_BODY_COUNT = 200;
+const RESOURCE_HEADERS_DEADLINE_MS = 30_000;
 
 function tick() {
 	return new Promise((resolve) => {
@@ -28,33 +20,6 @@ function tick() {
 	});
 }
 
-function createOutboundHTTP() {
-	const state = {inFlight: 0, peakInFlight: 0, started: 0, pending: []};
-	const outboundHTTP = {
-		request: (request) => {
-			if (state.inFlight >= OUTBOUND_MAX_IN_FLIGHT_PER_SERVICE) {
-				throw new DesktopOutboundHTTPCapacityError(`in-flight request for ${request.serviceName}`);
-			}
-			state.inFlight += 1;
-			state.started += 1;
-			state.peakInFlight = Math.max(state.peakInFlight, state.inFlight);
-			return new Promise((resolve) => {
-				state.pending.push(() => {
-					state.inFlight -= 1;
-					resolve({
-						headers: {'content-type': 'image/webp'},
-						message: Readable.from([Buffer.from('avatar-bytes')]),
-						status: 200,
-						statusText: 'OK',
-						url: new URL(request.url),
-					});
-				});
-			});
-		},
-	};
-	return {outboundHTTP, state};
-}
-
 function avatarRequest(index, signal) {
 	return {
 		targetURL: `https://media.fluxer.app/avatars/${index}.webp`,
@@ -68,131 +33,17 @@ function avatarRequest(index, signal) {
 	};
 }
 
-async function pump(state, work) {
-	let finished = false;
-	const result = work.finally(() => {
-		finished = true;
-	});
-	result.catch(() => {});
-	while (!finished) {
-		for (const settle of state.pending.splice(0)) {
-			settle();
-		}
-		await tick();
-	}
-	return await result;
-}
-
-function readAvatar(client, index, signal) {
-	return client.fetch(avatarRequest(index, signal)).then(async (response) => ({
-		status: response.status,
-		text: await response.text(),
-	}));
-}
-
-describe('the local app proxy bounds its own concurrency instead of overrunning outbound HTTP', () => {
-	test('every avatar of a fully rendered chat resolves rather than tripping the outbound capacity cap', async () => {
-		const {outboundHTTP, state} = createOutboundHTTP();
-		const client = new DesktopLocalAppProxyClient({outboundHTTP, traffic: LocalAppProxyTraffic.REMOTE_RESOURCE});
-		const controller = new AbortController();
-
-		const fetches = [];
-		for (let index = 0; index < RENDERED_AVATAR_COUNT; index += 1) {
-			fetches.push(readAvatar(client, index, controller.signal));
-		}
-		const responses = await pump(state, Promise.all(fetches));
-
-		assert.equal(responses.length, RENDERED_AVATAR_COUNT);
-		assert.deepEqual(new Set(responses.map((response) => response.status)), new Set([200]));
-		assert.equal(state.started, RENDERED_AVATAR_COUNT);
-		assert.deepEqual(
-			responses.map((response) => response.text),
-			new Array(RENDERED_AVATAR_COUNT).fill('avatar-bytes'),
-		);
-	});
-
-	test('no more requests are in flight at any instant than the proxy concurrency limit', async () => {
-		const {outboundHTTP, state} = createOutboundHTTP();
-		const client = new DesktopLocalAppProxyClient({outboundHTTP, traffic: LocalAppProxyTraffic.REMOTE_RESOURCE});
-		const controller = new AbortController();
-
-		const fetches = [];
-		for (let index = 0; index < RENDERED_AVATAR_COUNT; index += 1) {
-			fetches.push(readAvatar(client, index, controller.signal));
-		}
-		await tick();
-		const peakBeforeAnyCompletion = state.peakInFlight;
-		await pump(state, Promise.all(fetches));
-
-		assert.equal(peakBeforeAnyCompletion, PROXY_MAX_CONCURRENT_REQUESTS);
-		assert.equal(state.peakInFlight, PROXY_MAX_CONCURRENT_REQUESTS);
-	});
-
-	test('an aborted waiter rejects with the abort reason and leaves its queue slot to the next waiter', async () => {
-		const {outboundHTTP, state} = createOutboundHTTP();
-		const client = new DesktopLocalAppProxyClient({outboundHTTP, traffic: LocalAppProxyTraffic.REMOTE_RESOURCE});
-		const held = new AbortController();
-
-		const holding = [];
-		for (let index = 0; index < PROXY_MAX_CONCURRENT_REQUESTS; index += 1) {
-			holding.push(readAvatar(client, index, held.signal));
-		}
-		await tick();
-		assert.equal(state.started, PROXY_MAX_CONCURRENT_REQUESTS);
-
-		const abandoned = new AbortController();
-		const abandonedFetch = client.fetch(avatarRequest(998, abandoned.signal));
-		const survivor = readAvatar(client, 999, held.signal);
-		const reason = new Error('the renderer navigated away');
-		abandoned.abort(reason);
-
-		await assert.rejects(abandonedFetch, (error) => error === reason);
-		assert.equal(state.started, PROXY_MAX_CONCURRENT_REQUESTS);
-
-		state.pending.shift()();
-		await tick();
-		await tick();
-		assert.equal(state.started, PROXY_MAX_CONCURRENT_REQUESTS + 1);
-
-		const settled = await pump(state, Promise.all([survivor, ...holding]));
-		assert.equal(settled[0].status, 200);
-		assert.equal(settled.length, PROXY_MAX_CONCURRENT_REQUESTS + 1);
-	});
-
-	test('the queue refuses a request past its bound rather than growing without limit', async () => {
-		const {outboundHTTP, state} = createOutboundHTTP();
-		const client = new DesktopLocalAppProxyClient({outboundHTTP, traffic: LocalAppProxyTraffic.REMOTE_RESOURCE});
-		const controller = new AbortController();
-
-		const accepted = [];
-		for (let index = 0; index < PROXY_MAX_CONCURRENT_REQUESTS + PROXY_MAX_QUEUED_REQUESTS; index += 1) {
-			accepted.push(readAvatar(client, index, controller.signal));
-		}
-
-		await assert.rejects(
-			client.fetch(avatarRequest(Number.MAX_SAFE_INTEGER, controller.signal)),
-			(error) => error.name === 'LocalAppProxyQueueOverflowError',
-		);
-
-		const settled = await pump(state, Promise.all(accepted));
-		assert.equal(settled.length, PROXY_MAX_CONCURRENT_REQUESTS + PROXY_MAX_QUEUED_REQUESTS);
-	});
-});
-
-const RESOURCE_HEADERS_DEADLINE_MS = 30_000;
-
-function createSilentUpstream() {
+function createUpstream({streamBodies}) {
 	const state = {started: [], aborted: 0};
 	const outboundHTTP = {
 		request: (request) =>
 			new Promise((resolve, reject) => {
 				const entry = {
 					url: request.url,
-					serviceName: request.serviceName,
 					answer: () =>
 						resolve({
 							headers: {'content-type': 'image/webp'},
-							message: Readable.from([Buffer.from('avatar-bytes')]),
+							message: streamBodies ? new PassThrough() : Readable.from([Buffer.from('avatar-bytes')]),
 							status: 200,
 							statusText: 'OK',
 							url: new URL(request.url),
@@ -212,80 +63,64 @@ function createSilentUpstream() {
 	return {outboundHTTP, state};
 }
 
-function apiRequest(signal) {
-	return {...avatarRequest(0, signal), targetURL: 'https://api.fluxer.app/v1/users/@me'};
-}
+describe('the local app proxy never queues a request behind another', () => {
+	test('every avatar of a fully rendered chat starts at once', async () => {
+		const {outboundHTTP, state} = createUpstream({streamBodies: false});
+		const client = new DesktopLocalAppProxyClient({outboundHTTP, traffic: LocalAppProxyTraffic.REMOTE_RESOURCE});
+		const renderer = new AbortController();
 
-describe('media that a slow or silent self-host never answers cannot wedge the local app proxy', () => {
-	test('avatars dropped while their upstream sends no headers free their permits at the headers deadline', async () => {
+		const fetches = [];
+		for (let index = 0; index < RENDERED_AVATAR_COUNT; index += 1) {
+			fetches.push(client.fetch(avatarRequest(index, renderer.signal)));
+		}
+		await tick();
+		assert.equal(state.started.length, RENDERED_AVATAR_COUNT);
+
+		for (const entry of state.started) {
+			entry.answer();
+		}
+		const texts = await Promise.all(fetches.map(async (fetch) => await (await fetch).text()));
+		assert.deepEqual(texts, new Array(RENDERED_AVATAR_COUNT).fill('avatar-bytes'));
+	});
+
+	test('a media request starts at once while earlier response bodies sit unread', async () => {
+		const {outboundHTTP, state} = createUpstream({streamBodies: true});
+		const client = new DesktopLocalAppProxyClient({outboundHTTP, traffic: LocalAppProxyTraffic.REMOTE_RESOURCE});
+		const renderer = new AbortController();
+
+		for (let index = 0; index < PARKED_MEDIA_BODY_COUNT; index += 1) {
+			const fetch = client.fetch(avatarRequest(index, renderer.signal));
+			await tick();
+			state.started.at(-1).answer();
+			assert.equal((await fetch).status, 200);
+		}
+
+		const next = client.fetch(avatarRequest(PARKED_MEDIA_BODY_COUNT, renderer.signal));
+		await tick();
+		assert.equal(state.started.length, PARKED_MEDIA_BODY_COUNT + 1);
+		state.started.at(-1).answer();
+		assert.equal((await next).status, 200);
+		renderer.abort();
+	});
+});
+
+describe('media that a slow or silent self-host never answers is cut off', () => {
+	test('a silent upstream is aborted at the headers deadline', async () => {
 		mock.timers.enable({apis: ['setTimeout']});
 		try {
-			const {outboundHTTP, state} = createSilentUpstream();
+			const {outboundHTTP, state} = createUpstream({streamBodies: false});
 			const client = new DesktopLocalAppProxyClient({outboundHTTP, traffic: LocalAppProxyTraffic.REMOTE_RESOURCE});
 			const renderer = new AbortController();
-			const dropped = [];
-			for (let index = 0; index < PROXY_MAX_CONCURRENT_REQUESTS; index += 1) {
-				const fetch = client.fetch(avatarRequest(index, renderer.signal));
-				fetch.catch(() => {});
-				dropped.push(fetch);
-			}
+			const silent = client.fetch(avatarRequest(0, renderer.signal));
+			silent.catch(() => {});
 			await tick();
-			const fresh = client.fetch(avatarRequest(1000, renderer.signal));
-			await tick();
-			assert.equal(state.started.length, PROXY_MAX_CONCURRENT_REQUESTS);
 
 			mock.timers.tick(RESOURCE_HEADERS_DEADLINE_MS);
-			await Promise.allSettled(dropped);
-			await tick();
 
-			assert.equal(state.aborted, PROXY_MAX_CONCURRENT_REQUESTS);
-			assert.equal(state.started.length, PROXY_MAX_CONCURRENT_REQUESTS + 1);
-			state.started.at(-1).answer();
-			const response = await fresh;
-			assert.equal(response.status, 200);
-			assert.equal(await response.text(), 'avatar-bytes');
+			await assert.rejects(silent, (error) => error.name === 'LocalAppProxyResponseHeadersDeadlineError');
+			assert.equal(state.aborted, 1);
 		} finally {
 			mock.timers.reset();
 		}
 	});
-
-	test('API requests keep flowing while every media permit is held by a silent upstream', async () => {
-		const {outboundHTTP, state} = createSilentUpstream();
-		const media = new DesktopLocalAppProxyClient({outboundHTTP, traffic: LocalAppProxyTraffic.REMOTE_RESOURCE});
-		const api = new DesktopLocalAppProxyClient({outboundHTTP, traffic: LocalAppProxyTraffic.API});
-		const renderer = new AbortController();
-		for (let index = 0; index < PROXY_MAX_CONCURRENT_REQUESTS * 2; index += 1) {
-			media.fetch(avatarRequest(index, renderer.signal)).catch(() => {});
-		}
-		await tick();
-		const apiCall = api.fetch(apiRequest(renderer.signal));
-		await tick();
-		const started = state.started.at(-1);
-		assert.equal(started.url, 'https://api.fluxer.app/v1/users/@me');
-		assert.equal(started.serviceName, LocalAppProxyTraffic.API);
-		started.answer();
-		assert.equal((await apiCall).status, 200);
-		renderer.abort();
-	});
-
-	test('the newest queued media request is admitted first so what is on screen loads before what scrolled away', async () => {
-		const {outboundHTTP, state} = createSilentUpstream();
-		const client = new DesktopLocalAppProxyClient({outboundHTTP, traffic: LocalAppProxyTraffic.REMOTE_RESOURCE});
-		const renderer = new AbortController();
-		const fetches = [];
-		for (let index = 0; index < PROXY_MAX_CONCURRENT_REQUESTS + 3; index += 1) {
-			const fetch = client.fetch(avatarRequest(index, renderer.signal));
-			fetch.catch(() => {});
-			fetches.push(fetch);
-		}
-		await tick();
-		state.started[0].answer();
-		await (await fetches[0]).text();
-		await tick();
-		assert.equal(
-			state.started.at(-1).url,
-			`https://media.fluxer.app/avatars/${PROXY_MAX_CONCURRENT_REQUESTS + 2}.webp`,
-		);
-		renderer.abort();
-	});
 });

fluxer_desktop/src/main/SelectedInstanceFetch.ts

@@ -15,7 +15,6 @@ const INSTANCE_RESPONSE_MAX_BYTES = 1024 * 1024;
 const INSTANCE_RESPONSE_MAX_CHUNKS = 1024;
 const INSTANCE_FETCH_TIMEOUT_MS = 15_000;
 
-const SELECTED_INSTANCE_SERVICE_NAME = 'desktop_selected_instance';
 const INSTANCE_RESPONSE_DESCRIPTION = 'Instance response';
 const RESET_CONTENT_STATUS = 205;
 
@@ -80,7 +79,6 @@ export class DesktopSelectedInstanceClient {
 			headers: request.headers ?? null,
 			method,
 			originTrust: DesktopOriginTrust.BOUND,
-			serviceName: SELECTED_INSTANCE_SERVICE_NAME,
 			signal: request.signal ?? null,
 			timeoutMs,
 			url: request.url,

fluxer_docs/src/content/docs/operator/configuration.mdx

@@ -332,7 +332,7 @@ All are optional.
 
 #### `FLUXER_TRUST_CLIENT_IP_HEADER`
 
-Default `false`. Whether the client-IP header is trusted. Compose forwards it from `.env` with a default of `true`, because the edge sets the header on every upstream hop. The API has no peer-address fallback, so `false` on `api` makes every non-exempt request a 403.
+Default `false`. Whether the client-IP header is trusted. Compose forwards it from `.env` with a default of `true`, because the edge sets the header on every upstream hop. With `false`, `api` ignores the header and uses the address of the peer that opened the connection. That suits an `api` that clients reach directly. Behind any proxy it makes every request look as though it came from the proxy.
 
 #### `FLUXER_CLIENT_IP_HEADER_NAME`
 
@@ -342,7 +342,7 @@ Default `x-forwarded-for`. Which header has the client IP. Read by `api`, `worke
 
 Default empty. Addresses and CIDR ranges exempt from IP bans. Comma separated. A bare IPv4 address exempts that address, a bare IPv6 address exempts its /64, and a CIDR range such as `2001:db8:1200::/56` exempts every address in it. Every entry must parse as an IP or a CIDR range or the API fails at boot. Compose forwards it from `.env`.
 
-The API returns 403 for any request whose client-IP header is missing, empty, or not a parsable address, and for every request while `FLUXER_TRUST_CLIENT_IP_HEADER` is `false`. The exceptions are `/_health`, `/webhooks/livekit`, `/test`, and the Bluesky client metadata and JWKS routes. The edge sets the header on every upstream hop, so a missing or unparsable header happens only in a layout that puts something other than the edge directly in front of `api`. A proxy in front of the edge that never sets the header passes the check, and every request then looks as though it came from the proxy.
+While `FLUXER_TRUST_CLIENT_IP_HEADER` is `true`, the API returns 403 for any request whose client-IP header is missing, empty, or not a parsable address. The exceptions are `/_health`, `/webhooks/livekit`, `/test`, and the Bluesky client metadata and JWKS routes. The edge sets the header on every upstream hop, so a missing or unparsable header happens only in a layout that puts something other than the edge directly in front of `api`. A proxy in front of the edge that never sets the header passes the check, and every request then looks as though it came from the proxy.
 
 #### `FLUXER_API_TRUSTED_CALLERS`
 
@@ -514,6 +514,8 @@ Every uploaded file lands in an S3-compatible object store, which the stack prov
 
 Default `http://localhost:3900`. The S3 API address. `media-proxy` defaults to empty instead. Compose forwards it from `.env` with a default of `http://seaweedfs:8333`, the bundled store.
 
+An endpoint with a bucket in its path, such as `https://s3.example.com/my-bucket` with `FLUXER_S3_FORCE_PATH_STYLE=true`, keeps every Fluxer bucket inside that one bucket. Each bucket name becomes a key prefix there, so uploads land under `fluxer-uploads/` in `my-bucket`.
+
 #### `FLUXER_S3_PUBLIC_ENDPOINT`
 
 No default. The host substituted into presigned URLs, which are the only place it appears. Compose forwards it from `.env`, and uses `FLUXER_S3_ENDPOINT` when it is unset, except on `app-proxy`, which gets only a value `.env` sets and adds it to its Content Security Policy. `admin` reads it to find the origin that report downloads are presigned against. See `FLUXER_S3_BUCKET_REPORTS`.

packages/ip_utils/src/ClientIp.ts

@@ -1,11 +1,11 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
-import type {ParsedIpAddress} from '@fluxer/ip_utils/src/IpAddress';
-import {parseIpAddress} from '@fluxer/ip_utils/src/IpAddress';
+import {canonicalizeIpAddress} from '@fluxer/ip_utils/src/IpAddress';
 
 interface ClientIpExtractionOptions {
 	trustClientIpHeader?: boolean;
 	clientIpHeaderName?: string;
+	peerAddress?: string | null;
 }
 
 export class MissingClientIpError extends Error {
@@ -13,7 +13,7 @@ export class MissingClientIpError extends Error {
 	readonly status = 403;
 
 	constructor() {
-		super('Client IP header is required');
+		super('Client IP is required');
 		this.name = 'MissingClientIpError';
 	}
 }
@@ -28,20 +28,12 @@ export function resolveClientIpHeaderName(clientIpHeaderName?: string): string {
 	return normalizeHeaderName(clientIpHeaderName ?? DEFAULT_CLIENT_IP_HEADER_NAME);
 }
 
-function parseClientIpHeaderValue(value: string | null): ParsedIpAddress | null {
-	if (value === null) {
-		return null;
-	}
-	return parseIpAddress(value.split(',', 1)[0]);
-}
-
 export function extractClientIp(req: Request, options?: ClientIpExtractionOptions): string | null {
 	if (!options?.trustClientIpHeader) {
-		return null;
+		return options?.peerAddress ? canonicalizeIpAddress(options.peerAddress) : null;
 	}
-	return (
-		parseClientIpHeaderValue(req.headers.get(resolveClientIpHeaderName(options.clientIpHeaderName)))?.normalized ?? null
-	);
+	const headerValue = req.headers.get(resolveClientIpHeaderName(options.clientIpHeaderName));
+	return headerValue === null ? null : canonicalizeIpAddress(headerValue.split(',', 1)[0]);
 }
 
 export function requireClientIp(req: Request, options?: ClientIpExtractionOptions): string {

packages/ip_utils/src/IpAddress.ts

@@ -102,6 +102,17 @@ export function parseIpAddress(value: string): ParsedIpAddress | null {
 	};
 }
 
+export function canonicalizeIpAddress(value: string): string | null {
+	const parsed = parseIpAddress(value);
+	if (!parsed) {
+		return null;
+	}
+	if (parsed.family === 'ipv4') {
+		return parsed.normalized;
+	}
+	return getIpv4MappedIpv6(expandIpv6ToGroups(parsed.normalized)) ?? parsed.normalized;
+}
+
 export function isValidIp(value: string): boolean {
 	return parseIpAddress(value) !== null;
 }

packages/ip_utils/src/__tests__/ClientIp.test.ts

@@ -50,18 +50,43 @@ describe('extractClientIp', () => {
 			'192.168.1.1',
 		);
 	});
-	it('returns null when trust is disabled', () => {
+	it('ignores the header and uses the peer address when trust is disabled', () => {
 		const request = new Request('http://example.com', {
 			headers: {'X-Forwarded-For': '192.168.1.1'},
 		});
-		expect(extractClientIp(request, {trustClientIpHeader: false})).toBeNull();
-		expect(extractClientIp(request)).toBeNull();
+		expect(extractClientIp(request, {trustClientIpHeader: false, peerAddress: '203.0.113.7'})).toBe('203.0.113.7');
+		expect(extractClientIp(request, {peerAddress: '203.0.113.7'})).toBe('203.0.113.7');
 	});
-	it('returns null when no options are provided', () => {
+	it('uses the peer address when trust is disabled and no header is sent', () => {
+		const request = new Request('http://example.com');
+		expect(extractClientIp(request, {trustClientIpHeader: false, peerAddress: '2001:db8::7'})).toBe('2001:db8::7');
+	});
+	it('unmaps an ipv4-mapped peer address', () => {
+		const request = new Request('http://example.com');
+		expect(extractClientIp(request, {peerAddress: '::ffff:203.0.113.7'})).toBe('203.0.113.7');
+	});
+	it('returns null when trust is disabled and there is no usable peer address', () => {
 		const request = new Request('http://example.com', {
 			headers: {'X-Forwarded-For': '192.168.1.1'},
 		});
 		expect(extractClientIp(request)).toBeNull();
+		expect(extractClientIp(request, {trustClientIpHeader: false, peerAddress: null})).toBeNull();
+		expect(extractClientIp(request, {trustClientIpHeader: false, peerAddress: 'not-an-ip'})).toBeNull();
+	});
+	it('ignores the peer address when the header is trusted', () => {
+		const request = new Request('http://example.com', {
+			headers: {'X-Forwarded-For': '198.51.100.4'},
+		});
+		expect(extractClientIp(request, {trustClientIpHeader: true, peerAddress: '10.0.0.2'})).toBe('198.51.100.4');
+		expect(
+			extractClientIp(new Request('http://example.com'), {trustClientIpHeader: true, peerAddress: '10.0.0.2'}),
+		).toBeNull();
+	});
+	it('unmaps an ipv4-mapped header value', () => {
+		const request = new Request('http://example.com', {
+			headers: {'X-Forwarded-For': '::ffff:198.51.100.4, 10.0.0.1'},
+		});
+		expect(extractClientIp(request, {trustClientIpHeader: true})).toBe('198.51.100.4');
 	});
 	it('returns null for missing or invalid headers', () => {
 		expect(extractClientIp(new Request('http://example.com'), {trustClientIpHeader: true})).toBeNull();
@@ -115,9 +140,16 @@ describe('requireClientIp', () => {
 		});
 		expect(requireClientIp(request, {trustClientIpHeader: true})).toBe('192.168.1.1');
 	});
+	it('returns the peer address when the header is not trusted', () => {
+		const request = new Request('http://example.com', {
+			headers: {'X-Forwarded-For': '192.168.1.1'},
+		});
+		expect(requireClientIp(request, {trustClientIpHeader: false, peerAddress: '203.0.113.7'})).toBe('203.0.113.7');
+	});
 	it('throws typed error when missing', () => {
 		const request = new Request('http://example.com');
 		expect(() => requireClientIp(request, {trustClientIpHeader: true})).toThrow(MissingClientIpError);
-		expect(() => requireClientIp(request, {trustClientIpHeader: true})).toThrow('Client IP header is required');
+		expect(() => requireClientIp(request, {trustClientIpHeader: true})).toThrow('Client IP is required');
+		expect(() => requireClientIp(request, {trustClientIpHeader: false})).toThrow(MissingClientIpError);
 	});
 });

packages/ip_utils/src/__tests__/IpAddress.test.ts

@@ -1,6 +1,7 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
 import {
+	canonicalizeIpAddress,
 	getIpNetworkKey,
 	getSameIpDecisionKey,
 	getSubnet,
@@ -92,6 +93,21 @@ describe('parseIpAddress', () => {
 	});
 });
 
+describe('canonicalizeIpAddress', () => {
+	it('keeps ipv4 and plain ipv6 in their normalised form', () => {
+		expect(canonicalizeIpAddress(' 203.0.113.5 ')).toBe('203.0.113.5');
+		expect(canonicalizeIpAddress('[2001:DB8::1]')).toBe('2001:db8::1');
+	});
+	it('unmaps ipv4-mapped ipv6 in dotted and hex form', () => {
+		expect(canonicalizeIpAddress('::ffff:203.0.113.5')).toBe('203.0.113.5');
+		expect(canonicalizeIpAddress('::ffff:cb00:7105')).toBe('203.0.113.5');
+	});
+	it('returns null for anything that is not an address', () => {
+		expect(canonicalizeIpAddress('')).toBeNull();
+		expect(canonicalizeIpAddress('not-an-ip')).toBeNull();
+	});
+});
+
 describe('isValidIp', () => {
 	it('accepts valid values', () => {
 		expect(isValidIp('192.168.1.1')).toBe(true);