diff --git a/scripts/services/docker/Dockerfile.security_best_practices_worker b/scripts/services/docker/Dockerfile.security_best_practices_worker index cb250dc12f..04baff781b 100644 --- a/scripts/services/docker/Dockerfile.security_best_practices_worker +++ b/scripts/services/docker/Dockerfile.security_best_practices_worker @@ -1,4 +1,4 @@ -ARG PVTR_VERSION=v0.23.2 +ARG PVTR_VERSION=v0.24.0 FROM alpine:3.21 AS core RUN apk add --no-cache wget tar unzip @@ -10,13 +10,13 @@ ARG PLATFORM=Linux_x86_64 RUN wget https://github.com/privateerproj/privateer/releases/download/v${VERSION}/privateer_${PLATFORM}.tar.gz RUN tar -xzf privateer_${PLATFORM}.tar.gz -FROM golang:1.26.3-alpine3.23 AS plugin +FROM golang:1.26.4-alpine3.22 AS plugin RUN apk add --no-cache make git WORKDIR /plugin -ARG PVTR_COMMIT=c7bd9538d64f7eaab94a05c9b5fd05458a387b1c +ARG PVTR_COMMIT=c1095c95a1b399ec63e4f4e2b7880b0ef55e604f ARG PVTR_VERSION # To run the latest version of the plugin, we need to use the latest commit of the pvtr-github-repo-scanner repository. -# Currently using v0.23.2: https://github.com/ossf/pvtr-github-repo-scanner/commit/c7bd9538d64f7eaab94a05c9b5fd05458a387b1c +# Currently using v0.24.0+2fixes: https://github.com/ossf/pvtr-github-repo-scanner/commit/c1095c95a1b399ec63e4f4e2b7880b0ef55e604f RUN git clone https://github.com/ossf/pvtr-github-repo-scanner.git && cd pvtr-github-repo-scanner && git checkout ${PVTR_COMMIT} RUN cd pvtr-github-repo-scanner && make binary && cp github-repo ../github-repo diff --git a/services/apps/security_best_practices_worker/example-config.yml b/services/apps/security_best_practices_worker/example-config.yml index f6619448b1..8b3b8df719 100644 --- a/services/apps/security_best_practices_worker/example-config.yml +++ b/services/apps/security_best_practices_worker/example-config.yml @@ -12,6 +12,6 @@ services: - Maturity Level 1 vars: - owner: $REPO_OWNER - repo: $REPO_NAME - token: $GITHUB_TOKEN + owner: '$REPO_OWNER' + repo: '$REPO_NAME' + token: '$GITHUB_TOKEN' diff --git a/services/apps/security_best_practices_worker/src/activities.ts b/services/apps/security_best_practices_worker/src/activities.ts index 15ef05952d..a0bc8587b3 100644 --- a/services/apps/security_best_practices_worker/src/activities.ts +++ b/services/apps/security_best_practices_worker/src/activities.ts @@ -1,5 +1,6 @@ import { findObsoleteRepos, + getCurrentTimeMs, getOSPSBaselineInsights, initializeTokenInfos, saveOSPSBaselineInsightsToDB, @@ -12,4 +13,5 @@ export { findObsoleteRepos, initializeTokenInfos, updateTokenInfos, + getCurrentTimeMs, } diff --git a/services/apps/security_best_practices_worker/src/activities/index.ts b/services/apps/security_best_practices_worker/src/activities/index.ts index e1d92a5712..1b230c18a0 100644 --- a/services/apps/security_best_practices_worker/src/activities/index.ts +++ b/services/apps/security_best_practices_worker/src/activities/index.ts @@ -44,25 +44,12 @@ export async function getOSPSBaselineInsights(repoUrl: string, token: string): P const combinedOutput = `${stdout}\n${stderr}` - if (combinedOutput.includes('403')) { - svc.log.warn('Detected 403 error in privateer output!') - throw ApplicationFailure.create({ - message: 'GitHub token rate-limited', - type: 'Token403Error', - }) - } + classifyTokenError(combinedOutput, 'privateer output') } catch (err) { svc.log.error(`Privateer run failed: ${err.message}`) - // check for 403 in captured output if available const output = `${err.stdout || ''}\n${err.stderr || ''}` - if (output.includes('403')) { - svc.log.warn('Detected 403 error in failed privateer output!') - throw ApplicationFailure.create({ - message: 'GitHub token rate-limited', - type: 'Token403Error', - }) - } + classifyTokenError(output, 'failed privateer output') throw err } @@ -196,6 +183,40 @@ export async function saveOSPSBaselineInsightsToRedis( await redisCache.set(key, JSON.stringify(insights), 60 * 60 * 24) // 1 day } +function classifyTokenError(output: string, source: string): void { + const failedAtMs = Date.now() + + if (output.includes('401 Unauthorized') || output.includes('401 Bad credentials')) { + svc.log.warn(`Detected 401 error in ${source} - token invalid or expired!`) + throw ApplicationFailure.create({ + message: 'GitHub token invalid or expired', + type: 'TokenAuthError', + nonRetryable: true, + details: [failedAtMs], + }) + } + + // Word-boundary match so unrelated numbers don't get misclassified as HTTP 429/403. + const has403 = /\b403\b/.test(output) + const has429 = /\b429\b/.test(output) + if (!has403 && !has429) return + + // 429 is always rate-limit. 403 with rate-limit body is rate-limit. 403 without is a + // permission problem (SAML, missing scopes) — log it but don't throw; the child workflow's + // retry policy will exhaust retries and the repo will be deferred to the next scheduled run. + const isRateLimit = has429 || /rate limit|rate_limit|secondary rate/i.test(output) + if (isRateLimit) { + svc.log.warn(`Detected rate-limit in ${source} - token rate-limited!`) + throw ApplicationFailure.create({ + message: 'GitHub token rate-limited', + type: 'Token403Error', + nonRetryable: true, + details: [failedAtMs], + }) + } + svc.log.warn(`Detected 403 permission error in ${source} - token may lack access to this repo`) +} + function computeRunDuration(start: string | undefined, end: string | undefined): string { if (!start || !end) return '' const startMs = new Date(start).getTime() @@ -258,13 +279,13 @@ async function runBinary( resolve({ stdout, stderr }) } else { const truncated = (s: string) => (s.length > 500 ? s.slice(0, 500) + '…' : s) - const truncStdout = truncated(stdout) - const truncStderr = truncated(stderr) + // Attach full stdout/stderr so classifyTokenError sees rate-limit markers that may + // appear after the truncation cut. Message is still truncated for log readability. const err = Object.assign( new Error( - `Binary exited with code ${code}\nStderr:\n${truncStderr}\nStdout:\n${truncStdout}`, + `Binary exited with code ${code}\nStderr:\n${truncated(stderr)}\nStdout:\n${truncated(stdout)}`, ), - { stdout: truncStdout, stderr: truncStderr }, + { stdout, stderr }, ) reject(err) } @@ -276,20 +297,45 @@ export async function initializeTokenInfos(): Promise { const redisCache = new RedisCache(`osps-baseline-insights`, svc.redis, svc.log) const tokenInfosInRedis = await redisCache.get('tokenInfos') - - if (tokenInfosInRedis) { - return JSON.parse(tokenInfosInRedis) - } - - return process.env['CROWD_GITHUB_PERSONAL_ACCESS_TOKENS'].split(',').map((token) => ({ - token, - inUse: false, - lastUsed: new Date(), - isRateLimited: false, - })) + const cached: ITokenInfo[] = tokenInfosInRedis ? JSON.parse(tokenInfosInRedis) : [] + const cachedByToken = new Map(cached.map((t) => [t.token, t])) + + // Env var is authoritative for pool membership so PAT rotation is picked up on next run: + // a new token in CROWD_GITHUB_PERSONAL_ACCESS_TOKENS joins the pool fresh, and a removed + // token drops out even if its cached entry is still in Redis. + const envTokens = process.env['CROWD_GITHUB_PERSONAL_ACCESS_TOKENS'].split(',') + + return envTokens.map((token) => { + const t = cachedByToken.get(token) + if (t) { + return { + ...t, + // Reset inUse — workflow processes may have crashed leaving stale in-use flags. + inUse: false, + // Backward compat: legacy entries have isRateLimited=true with no rateLimitedAt timestamp. + // Without a timestamp the 1-hour expiry can't apply and the token would be stuck forever. + // Clear stale rate-limits that lack a timestamp so they can be retried on this run. + isRateLimited: t.isRateLimited && !!t.rateLimitedAt, + rateLimitedAt: t.isRateLimited && t.rateLimitedAt ? t.rateLimitedAt : undefined, + } + } + return { + token, + inUse: false, + lastUsed: new Date(), + isRateLimited: false, + } + }) } export async function updateTokenInfos(tokenInfos: ITokenInfo[]): Promise { const redisCache = new RedisCache(`osps-baseline-insights`, svc.redis, svc.log) await redisCache.set('tokenInfos', JSON.stringify(tokenInfos), 60 * 60 * 24) // 1 day } + +// Wall-clock time via activity — workflow code can't call Date.now() (determinism rule), +// but rate-limit cooldowns need real elapsed time to expire mid-run rather than only at +// the next continueAsNew batch boundary. +export async function getCurrentTimeMs(): Promise { + return Date.now() +} diff --git a/services/apps/security_best_practices_worker/src/types.ts b/services/apps/security_best_practices_worker/src/types.ts index e50a44a4f2..7945eb9fc1 100644 --- a/services/apps/security_best_practices_worker/src/types.ts +++ b/services/apps/security_best_practices_worker/src/types.ts @@ -52,6 +52,10 @@ export interface ITriggerSecurityInsightsCheckForReposParams { export interface ITokenInfo { token: string inUse: boolean - lastUsed: Date + // Date at initialization time; becomes an ISO string after JSON round-trip through Redis + // and Temporal payloads, so callers must wrap in `new Date()` before comparing. + lastUsed: Date | string isRateLimited: boolean + rateLimitedAt?: string // ISO timestamp; used to auto-reset after 1 hour + isInvalid?: boolean // 401 auth failure; recover by rotating the PAT out of CROWD_GITHUB_PERSONAL_ACCESS_TOKENS } diff --git a/services/apps/security_best_practices_worker/src/workflows/triggerSecurityInsightsCheckForRepos.ts b/services/apps/security_best_practices_worker/src/workflows/triggerSecurityInsightsCheckForRepos.ts index edfe0f048c..2d16068856 100644 --- a/services/apps/security_best_practices_worker/src/workflows/triggerSecurityInsightsCheckForRepos.ts +++ b/services/apps/security_best_practices_worker/src/workflows/triggerSecurityInsightsCheckForRepos.ts @@ -13,12 +13,13 @@ import { ITokenInfo, ITriggerSecurityInsightsCheckForReposParams } from '../type import { upsertOSPSBaselineSecurityInsights } from './upsertOSPSBaselineSecurityInsights' -const { findObsoleteRepos, initializeTokenInfos, updateTokenInfos } = proxyActivities< - typeof activities ->({ - startToCloseTimeout: '5 minutes', - retry: { maximumAttempts: 3, backoffCoefficient: 3 }, -}) +const { findObsoleteRepos, initializeTokenInfos, updateTokenInfos, getCurrentTimeMs } = + proxyActivities({ + startToCloseTimeout: '5 minutes', + retry: { maximumAttempts: 3, backoffCoefficient: 3 }, + }) + +const ONE_HOUR_MS = 60 * 60 * 1000 export async function triggerSecurityInsightsCheckForRepos( args: ITriggerSecurityInsightsCheckForReposParams, @@ -26,12 +27,34 @@ export async function triggerSecurityInsightsCheckForRepos( const REPOS_OBSOLETE_AFTER_SECONDS = 30 * 24 * 60 * 60 // 30 days const LIMIT_REPOS_TO_CHECK_PER_RUN = 100 const MAX_PARALLEL_CHILDREN = 5 - const MAX_TOKEN_ATTEMPTS = 5 const info = workflowInfo() const failedRepoUrls = args?.failedRepoUrls || [] + // Wall-clock time comes from the getCurrentTimeMs activity — workflow code can't call + // Date.now() (Temporal determinism rule). Refreshed each outer-loop iteration so rate-limit + // cooldowns can expire mid-run instead of only at continueAsNew boundaries. + let currentTimeMs = await getCurrentTimeMs() + + // Token state is persisted in Redis via updateTokenInfos so isRateLimited (with rateLimitedAt + // for expiry) and isInvalid survive across continueAsNew batches. const tokenInfos: ITokenInfo[] = await initializeTokenInfos() + + // Empty pool means CROWD_GITHUB_PERSONAL_ACCESS_TOKENS is misconfigured. Without this guard + // an empty array would satisfy `tokenInfos.every(t => t.isInvalid)` and silently mark every + // obsolete repo as failed for the continueAsNew chain, hiding the config error. + if (tokenInfos.length === 0) { + throw ApplicationFailure.create({ + message: 'No GitHub tokens configured (CROWD_GITHUB_PERSONAL_ACCESS_TOKENS is empty)', + type: 'TokenPoolEmpty', + nonRetryable: true, + }) + } + + // Scale attempts to pool size so a repo isn't marked failed while untried tokens remain + // (e.g. 6 PATs where the first 5 attempts all 403 on different tokens). + const MAX_TOKEN_ATTEMPTS = Math.max(5, tokenInfos.length) + const repos = await findObsoleteRepos( REPOS_OBSOLETE_AFTER_SECONDS, failedRepoUrls, @@ -48,10 +71,28 @@ export async function triggerSecurityInsightsCheckForRepos( async function processRepo(repo: (typeof repos)[0]): Promise { let attempts = 0 while (attempts < MAX_TOKEN_ATTEMPTS) { - const tokenInfo = await getNextToken(tokenInfos) + // Refresh wall-clock time on retries only. First attempt intentionally reuses the + // outer-loop's currentTimeMs so processRepo runs synchronously up to + // `tokenInfo.inUse = true` — otherwise the outer loop can oversubscribe + // MAX_PARALLEL_CHILDREN tasks against a smaller token pool before any inUse mark lands. + if (attempts > 0) { + currentTimeMs = await getCurrentTimeMs() + } + const tokenInfo = getNextToken(tokenInfos, currentTimeMs) + if (!tokenInfo) { + // No usable token — requeue if any may recover (rate-limited tokens expire after 1h + // or free up from concurrent children); fail if all are permanently invalid. + const allPermanentlyInvalid = tokenInfos.every((t) => t.isInvalid) + if (allPermanentlyInvalid) { + console.error(`No usable tokens for repo ${repo.repoUrl}, skipping`) + failedRepoUrls.push(repo.repoUrl) + } else { + queue.unshift(repo) + } + break + } const token = tokenInfo.token - - await acquireToken(tokenInfos, token) + tokenInfo.inUse = true try { await executeChild(upsertOSPSBaselineSecurityInsights, { @@ -63,7 +104,7 @@ export async function triggerSecurityInsightsCheckForRepos( initialInterval: 2000, backoffCoefficient: 2, maximumInterval: 30000, - nonRetryableErrorTypes: ['Token403Error'], + nonRetryableErrorTypes: ['Token403Error', 'TokenAuthError'], }, args: [ { @@ -76,20 +117,41 @@ export async function triggerSecurityInsightsCheckForRepos( }) return // success } catch (error) { - if (error instanceof ApplicationFailure && error.type === 'Token403Error') { + // executeChild rejects with ChildWorkflowFailure wrapping ActivityFailure wrapping + // ApplicationFailure — traverse the cause chain to find the root ApplicationFailure. + const appFailure = unwrapApplicationFailure(error) + if (appFailure?.type === 'Token403Error') { tokenInfo.isRateLimited = true + tokenInfo.rateLimitedAt = + extractFailureTimestamp(appFailure) ?? new Date(currentTimeMs).toISOString() + attempts++ + continue // retry with a different token + } else if (appFailure?.type === 'TokenAuthError') { + console.error( + `Token invalid/expired for repo ${repo.repoUrl}, permanently excluding token`, + ) + tokenInfo.isInvalid = true attempts++ continue // retry with a different token } else { console.error(`Failed to process repo ${repo.repoUrl}:`, error) - // we retried this error using the retry policy (because it's not a token non-retryable error) - // but it failed in all retries. - // to proceed with processing, we don't wanna try this repo again in this run failedRepoUrls.push(repo.repoUrl) break } } finally { - await releaseToken(tokenInfos, tokenInfo.token) + tokenInfo.inUse = false + tokenInfo.lastUsed = new Date(currentTimeMs) + } + } + + if (attempts >= MAX_TOKEN_ATTEMPTS) { + const allPermanentlyInvalid = tokenInfos.every((t) => t.isInvalid) + if (!allPermanentlyInvalid) { + console.warn(`Exhausted token attempts for repo ${repo.repoUrl}, requeuing`) + queue.unshift(repo) + } else { + console.error(`All tokens invalid for repo ${repo.repoUrl}, skipping`) + failedRepoUrls.push(repo.repoUrl) } } } @@ -102,8 +164,18 @@ export async function triggerSecurityInsightsCheckForRepos( * it will be removed from the activeTasks array and the next task will be started. * This way we don't need to wait for all tasks to finish before starting new ones. */ + let deferred = false while (queue.length > 0 || activeTasks.length > 0) { + // Refresh wall-clock time each iteration so mid-run rate-limit cooldowns can expire + // without waiting for the next continueAsNew batch. + currentTimeMs = await getCurrentTimeMs() while (queue.length > 0 && activeTasks.length < MAX_PARALLEL_CHILDREN) { + // Only start a task if a token is currently free — avoids false "no token" failures + // when all tokens are temporarily held by other concurrent tasks. + refreshExpiredRateLimits(tokenInfos, currentTimeMs) + const hasFreeToken = tokenInfos.some((t) => !t.isInvalid && !t.isRateLimited && !t.inUse) + if (!hasFreeToken) break + const repo = queue.shift() const task = processRepo(repo).finally(() => { const index = activeTasks.indexOf(task) @@ -112,21 +184,69 @@ export async function triggerSecurityInsightsCheckForRepos( activeTasks.push(task) } - // Wait for any one to finish before refilling if (activeTasks.length > 0) { await Promise.race(activeTasks) + } else if (queue.length > 0) { + // No active tasks and no free tokens. If every token is permanently invalid, deferring + // would loop forever across scheduled runs — fail the remaining repos so ops sees the + // problem instead of a silent stall. + const allPermanentlyInvalid = tokenInfos.every((t) => t.isInvalid) + if (allPermanentlyInvalid) { + console.error( + `All tokens permanently invalid; marking ${queue.length} remaining repos as failed`, + ) + for (const repo of queue) failedRepoUrls.push(repo.repoUrl) + queue.length = 0 + break + } + // Otherwise defer: remaining repos are still obsolete and findObsoleteRepos will pick + // them up on the next scheduled run (daily cron `0 8 * * *`). We don't `sleep()` here + // to the earliest cooldown because the 30-day obsolescence window makes a ~23h delay + // negligible, and holding a worker slot idle for hours per rate-limit event isn't worth + // the cost. `rateLimitedAt` persists in Redis so the 1h expiry still applies on resume. + console.warn( + `No tokens available; deferring ${queue.length} repos to next batch (they will be re-fetched)`, + ) + deferred = true + break } } + // Persist token state to Redis so isRateLimited (with rateLimitedAt for expiry) and isInvalid + // survive the continueAsNew handoff. await updateTokenInfos(tokenInfos) + if (deferred) { + // All tokens rate-limited or invalid and repos still pending. Calling continueAsNew here + // would spin: findObsoleteRepos returns the same repos, they defer again, etc. + // Return instead — the daily schedule (scheduleCheckReposWithObsoleteSecurityInsights) + // will re-trigger the workflow; rate-limit timestamps persist in Redis so recovery is + // resumable once the 1h window elapses. + return + } + await continueAsNew({ failedRepoUrls, }) } -async function getNextToken(tokenInfos: ITokenInfo[]): Promise { - const usableTokenInfos = tokenInfos.filter((token) => !token.inUse && !token.isRateLimited) +function refreshExpiredRateLimits(tokenInfos: ITokenInfo[], nowMs: number): void { + for (const t of tokenInfos) { + if (!t.isRateLimited) continue + // Clear rate-limit when the timestamp is missing, unparseable, or older than 1h. + // Guarding against NaN avoids permanently wedging a token due to malformed cache data. + const rateMs = t.rateLimitedAt ? new Date(t.rateLimitedAt).getTime() : NaN + if (!Number.isFinite(rateMs) || nowMs - rateMs > ONE_HOUR_MS) { + t.isRateLimited = false + t.rateLimitedAt = undefined + } + } +} + +function getNextToken(tokenInfos: ITokenInfo[], nowMs: number): ITokenInfo | null { + refreshExpiredRateLimits(tokenInfos, nowMs) + + const usableTokenInfos = tokenInfos.filter((t) => !t.inUse && !t.isRateLimited && !t.isInvalid) // sort usable tokens by last used date from oldest to newest const sortedTokenInfos = usableTokenInfos.sort((a, b) => { @@ -135,24 +255,22 @@ async function getNextToken(tokenInfos: ITokenInfo[]): Promise { return aTime - bTime }) - if (sortedTokenInfos.length === 0) { - throw new Error('No usable tokens available') - } - - return sortedTokenInfos[0] + return sortedTokenInfos[0] ?? null } -async function releaseToken(tokenInfos: ITokenInfo[], token: string): Promise { - const tokenInfo = tokenInfos.find((tokenInfo) => tokenInfo.token === token) - if (tokenInfo) { - tokenInfo.inUse = false - tokenInfo.lastUsed = new Date() +function unwrapApplicationFailure(error: unknown): ApplicationFailure | null { + let e: unknown = error + while (e) { + if (e instanceof ApplicationFailure) return e + e = (e as { cause?: unknown }).cause } + return null } -async function acquireToken(tokenInfos: ITokenInfo[], token: string): Promise { - const tokenInfo = tokenInfos.find((tokenInfo) => tokenInfo.token === token) - if (tokenInfo) { - tokenInfo.inUse = true - } +// Activity records the wall-clock failure time via ApplicationFailure.details[0] (ms). +function extractFailureTimestamp(appFailure: ApplicationFailure): string | null { + const details = appFailure.details as unknown[] | undefined + const ts = details?.[0] + if (typeof ts !== 'number' || !Number.isFinite(ts)) return null + return new Date(ts).toISOString() } diff --git a/services/apps/security_best_practices_worker/src/workflows/upsertOSPSBaselineSecurityInsights.ts b/services/apps/security_best_practices_worker/src/workflows/upsertOSPSBaselineSecurityInsights.ts index bab2a5ff16..eddb74c74f 100644 --- a/services/apps/security_best_practices_worker/src/workflows/upsertOSPSBaselineSecurityInsights.ts +++ b/services/apps/security_best_practices_worker/src/workflows/upsertOSPSBaselineSecurityInsights.ts @@ -6,7 +6,7 @@ import { IUpsertOSPSBaselineSecurityInsightsParams } from '../types' const { getOSPSBaselineInsights, saveOSPSBaselineInsightsToDB } = proxyActivities< typeof activities >({ - startToCloseTimeout: '5 minutes', + startToCloseTimeout: '15 minutes', retry: { maximumAttempts: 5, initialInterval: 2 * 1000, backoffCoefficient: 1 }, })