Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions requirements/cli.md
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,9 @@
identity").
- Both real and fabricated users are persisted across runtime restarts: the same token resolves to
the same user on every subsequent request.
- A new token that resolves to a real account the runtime already knows joins that user rather than
creating another one, and the user's earlier tokens keep resolving (the CLI's OAuth login rotates
its token every hour).
- **Proxy password**: the adopted real proxy password (when known) is forwarded as
`APIFY_PROXY_PASSWORD` into every Actor run container, with the runtime's own
`APIFY_PROXY_PASSWORD` environment variable taking precedence when set (see `actor-driver.md`).
Expand Down
3 changes: 3 additions & 0 deletions src/api/auth.ts
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,8 @@ import { sendError } from './envelope.js';
declare module 'express-serve-static-core' {
interface Request {
user?: UserRecord;
/** The token this request presented, as distinct from `user.token`, the user's latest one. */
token?: string;
}
}

Expand Down Expand Up @@ -40,6 +42,7 @@ export function auth(): RequestHandler {
// already produces the same 500/`internal-error` envelope and (unlike a local catch) logs the real
// error via `console.error` first.
req.user = await getOrCreateUserForToken(token);
req.token = token;
next();
};
}
Expand Down
6 changes: 3 additions & 3 deletions src/services/api-fallback.ts
Original file line number Diff line number Diff line change
Expand Up @@ -147,7 +147,7 @@ export interface LocalError {
* all, e.g. one outside `/v2` entirely). All HTTP methods are eligible once these hold, writes included.
*
* Replay is `<upstreamApiBaseUrl()><req.originalUrl>` (byte-exact, percent-encoding intact), the
* caller's own presented token (`req.user.token`, unconditionally - see `services/users.ts`) as the only
* caller's own presented token (`req.token`, never `req.user.token`, which is the user's latest) as the only
* `Authorization` header, `content-type`/`accept` forwarded when the inbound request carried them,
* nothing else. One attempt, redirects followed, a 30s timeout. Only a final `2xx` is relayed: status
* and body unchanged, headers minus the hop-by-hop exclusion set below. `Set-Cookie` is always relayed
Expand Down Expand Up @@ -183,10 +183,10 @@ export async function attemptFallback(req: Request, res: Response, localError: L
if (!toggleOn) return false;

if (!isEligibleUpstreamPath(req.originalUrl)) return false;
if (!req.user) return false;
if (!req.user || !req.token) return false;

const method = req.method.toUpperCase();
const headers: Record<string, string> = { authorization: `Bearer ${req.user.token}` };
const headers: Record<string, string> = { authorization: `Bearer ${req.token}` };
const contentType = req.header('content-type');
if (contentType) headers['content-type'] = contentType;
const accept = req.header('accept');
Expand Down
47 changes: 38 additions & 9 deletions src/services/users.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@ import type { UserRecord } from '../storage/entities.js';
import { getRegistries } from '../storage/registries.js';
import { KeyedMutex } from '../storage/mutex.js';
import { isApifyProxyEnabled } from './apify-proxy.js';
import { fetchRealIdentity, upstreamApiBaseUrl } from './identity-resolution.js';
import { fetchRealIdentity, type RealIdentity, upstreamApiBaseUrl } from './identity-resolution.js';

/** Serialises "resolve token -> create-if-missing" per token, so two concurrent first-requests for the
* same brand-new token can never both pass the not-found check and mint two users - the same
Expand Down Expand Up @@ -43,13 +43,34 @@ async function fabricateUser(token: string): Promise<UserRecord> {
id: `0000000000000000${n}`,
username: `local-user-${n}`,
token,
tokens: [token],
createdAt: new Date().toISOString(),
};
}

/** Enough for a run container started with a token that has since rotated hourly for days. */
const MAX_REMEMBERED_TOKENS = 100;

function tokensOf(user: UserRecord): string[] {
return user.tokens ?? [user.token];
}

async function findUserByToken(token: string): Promise<UserRecord | null> {
const all = await getRegistries().users.list();
return all.find((user) => user.token === token) ?? null;
return all.find((user) => tokensOf(user).includes(token)) ?? null;
}

/** A token that resolved to an account the runtime already knows (an OAuth login's token rotates every
* hour) joins that user: its earlier tokens keep resolving, so a run container or a restart does not
* lose the account, and the new one becomes the token handed out from now on. */
function withToken(existing: UserRecord, token: string, identity: RealIdentity): UserRecord {
return {
...existing,
username: identity.username,
token,
tokens: [...tokensOf(existing).filter((known) => known !== token), token].slice(-MAX_REMEMBERED_TOKENS),
proxyPassword: identity.proxyPassword ?? existing.proxyPassword,
};
}

/**
Expand Down Expand Up @@ -90,19 +111,27 @@ export async function getOrCreateUserForToken(token: string): Promise<UserRecord
const identity = await fetchRealIdentity(token);
let user: UserRecord;
if (identity) {
user = {
id: identity.id,
username: identity.username,
token,
createdAt: new Date().toISOString(),
proxyPassword: identity.proxyPassword,
};
const existing = await getRegistries().users.get(identity.id);
user = existing
? withToken(existing, token, identity)
: {
id: identity.id,
username: identity.username,
token,
tokens: [token],
createdAt: new Date().toISOString(),
proxyPassword: identity.proxyPassword,
};
} else {
console.log(`could not resolve token against ${upstreamApiBaseUrl()}, using local identity`);
user = await fabricateUser(token);
}

await getRegistries().users.set(user.id, user);
// Requests still carrying an earlier token of this user get the updated record too.
for (const [knownToken, knownUser] of userByToken) {
if (knownUser.id === user.id) userByToken.set(knownToken, user);
}
userByToken.set(token, user);
return user;
});
Expand Down
3 changes: 3 additions & 0 deletions src/storage/entities.ts
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,10 @@ export interface DatasetSchema {
export interface UserRecord {
id: string;
username: string;
/** The token seen last, handed to new runs, schedules and standby links. */
token: string;
/** Every token this user was seen with, oldest first; absent on records written before tokens rotated. */
tokens?: string[];
createdAt: string;
/** Real Apify Proxy password for this account, harvested from the upstream `/v2/users/me` response
* at creation time when the token resolved against the real platform (see
Expand Down
27 changes: 27 additions & 0 deletions test/integration/api-fallback.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -892,6 +892,33 @@ describe('api-fallback: eligibility, relay, and fail-closed behaviour', () => {
}
});

it("relays the token the caller sent, not the user's latest one, when a user has several tokens", async () => {
// Both tokens resolve to the same real account, so the second joins the first's user and becomes
// its latest token (`services/users.ts`).
const identity = await startStubUpstream(() => ({
status: 200,
body: { data: { id: 'rotated-real-id', username: 'rotated-real-username' } },
}));
process.env.APIFY_UPSTREAM_API_BASE_URL = identity.baseUrl;
try {
await warmUpIdentity(server.baseUrl, 'rotated-token-earlier');
await warmUpIdentity(server.baseUrl, 'rotated-token-latest');
} finally {
await identity.close();
}
expect((await getRegistries().users.get('rotated-real-id'))?.token).toBe('rotated-token-latest');

const stub = await startStubUpstream(fixedOkResponse('rotated-token-check'));
process.env.APIFY_UPSTREAM_API_BASE_URL = stub.baseUrl;
try {
await call('get', '/v2/totally-made-up-path', { token: 'rotated-token-earlier' });
expect(stub.hitCount()).toBe(1);
expect(stub.requests()[0]?.headers.authorization).toBe('Bearer rotated-token-earlier');
} finally {
await stub.close();
}
});

it('a request with no token at all never reaches the stub (rejected by auth() first)', async () => {
const stub = await startStubUpstream(fixedOkResponse('no-token-check'));
process.env.APIFY_UPSTREAM_API_BASE_URL = stub.baseUrl;
Expand Down
34 changes: 34 additions & 0 deletions test/integration/identity-resolution.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -175,6 +175,40 @@ describe('identity resolution against the real platform (stubbed upstream)', ()
await stub.close();
}
});

it('a new token of an already-known account joins that user, and its earlier tokens keep resolving after a restart', async () => {
const stub = await startStubUpstream({
data: {
id: 'rotating-real-id',
username: 'rotating-real-username',
proxy: { password: 'rotating-password' },
},
});
process.env.APIFY_UPSTREAM_API_BASE_URL = stub.baseUrl;
try {
const first = clientWithToken(server.baseUrl, 'rotating-token-1');
const second = clientWithToken(server.baseUrl, 'rotating-token-2');

expect(((await first.user('me').get()) as unknown as UserMeResponse).id).toBe('rotating-real-id');
expect(((await second.user('me').get()) as unknown as UserMeResponse).id).toBe('rotating-real-id');
expect(stub.hitCount()).toBe(2);

// One record: the latest token is the one handed out, the earlier one is still known.
const all = (await getRegistries().users.list()).filter((user) => user.id === 'rotating-real-id');
expect(all).toHaveLength(1);
expect(all[0]?.token).toBe('rotating-token-2');
expect(all[0]?.tokens).toEqual(['rotating-token-1', 'rotating-token-2']);

const { resetUsersForTests } = await import('../../src/services/users.js');
resetUsersForTests();

// A run container started with the first token is still that user, with no re-probe.
expect(((await first.user('me').get()) as unknown as UserMeResponse).id).toBe('rotating-real-id');
expect(stub.hitCount()).toBe(2);
} finally {
await stub.close();
}
});
});

describe('harvested proxy password flows into Actor run containers, per user', () => {
Expand Down
Loading