Add an OAuth 2.1 authorization server for external agents

.env.template

@@ -81,3 +81,13 @@ CHROMATIC_PROJECT_TOKEN=chpt_sample_secret
 # THEME
 OPS_DARK_THEME_ENABLED=false
 OPS_CODE_BLOCK_MEMORY_LIMIT_IN_MB=256
+
+# EXTERNAL OAUTH
+OPS_OAUTH_ENABLED=false
+OPS_OAUTH_ISSUER_URL=http://localhost:3000
+OPS_MCP_RESOURCE_URL=
+OPS_OAUTH_RS_CLIENT_SECRET=
+OPS_OAUTH_REFRESH_TOKEN_TTL_DAYS=30
+OPS_OAUTH_ACCESS_TOKEN_TTL_SECONDS=900
+OPS_OAUTH_EXCHANGE_TOKEN_TTL_SECONDS=300
+OPS_OAUTH_SIGNING_KEY_PEM_PATH=

.gitignore

@@ -10,6 +10,7 @@ node_modules
 /tmp
 /.nx
 /cache
+**/cache/codes/
 /packages/ui-components/storybook-static
 
 

packages/server/api/src/app/app.ts

@@ -53,6 +53,9 @@ import { formModule } from './flows/flow/form/form.module';
 import { folderModule } from './flows/folder/folder.module';
 import { triggerEventModule } from './flows/trigger-events/trigger-event.module';
 import { systemJobsSchedule } from './helper/system-jobs';
+import { oauthConfig } from './oauth/config/oauth-config';
+import { registerOAuthCleanupHandler } from './oauth/oauth-cleanup-job';
+import { oauthModule } from './oauth/oauth.module';
 import { organizationModule } from './organization/organization.module';
 import { projectModule } from './project/project-module';
 import { slackInteractionModule } from './slack/slack-interaction-module';
@@ -225,6 +228,14 @@ export const setupApp = async (
   await app.register(blockVariableModule);
   await app.register(benchmarkModule);
 
+  // Unconditional: the schedule lives in Redis and survives OAuth being turned off, so
+  // the handler must exist even then. It no-ops while disabled.
+  registerOAuthCleanupHandler();
+
+  if (oauthConfig.isEnabled()) {
+    await app.register(oauthModule);
+  }
+
   app.get(
     '/redirect',
     async (

packages/server/api/src/app/authentication/context/access-token-manager.ts

@@ -12,8 +12,14 @@ import {
   WorkerMachineType,
   WorkerPrincipal,
 } from '@openops/shared';
+import jwtLibrary from 'jsonwebtoken';
 import { nanoid } from 'nanoid';
-import { jwtUtils } from '../../helper/jwt-utils';
+import { JwtSignAlgorithm, jwtUtils } from '../../helper/jwt-utils';
+import { OAuthError } from '../../oauth/common/oauth-errors';
+import { oauthConfig } from '../../oauth/config/oauth-config';
+import { buildOAuthServicePrincipal } from '../../oauth/projects/service-principal';
+import { OAuthAccessTokenClaims } from '../../oauth/storage/oauth-model';
+import { signingKeyService } from '../../oauth/tokens/signing-key.service';
 
 const openOpsRefreshTokenLifetimeSeconds =
   (system.getNumber(AppSystemProp.JWT_TOKEN_LIFETIME_HOURS) ?? 168) * 3600;
@@ -111,6 +117,10 @@ export const accessTokenManager = {
   },
 
   async extractPrincipal(token: string): Promise<Principal> {
+    if (isOAuthIssuedToken(token)) {
+      return extractOAuthPrincipal(token);
+    }
+
     const secret = await jwtUtils.getJwtSecret();
 
     try {
@@ -133,6 +143,51 @@ export const accessTokenManager = {
   },
 };
 
+function isOAuthIssuedToken(token: string): boolean {
+  return (
+    jwtLibrary.decode(token, { complete: true })?.header?.alg ===
+    JwtSignAlgorithm.RS256
+  );
+}
+
+async function extractOAuthPrincipal(token: string): Promise<Principal> {
+  const invalidToken = new ApplicationError({
+    code: ErrorCode.INVALID_BEARER_TOKEN,
+    params: {
+      message: 'invalid access token',
+    },
+  });
+
+  if (!oauthConfig.isEnabled()) {
+    throw invalidToken;
+  }
+
+  try {
+    const claims = await signingKeyService.verifyAccessToken(
+      token,
+      oauthConfig.getApiAudience(),
+    );
+
+    return await buildOAuthServicePrincipal(
+      claims as unknown as OAuthAccessTokenClaims,
+    );
+  } catch (error) {
+    // Only a verdict about the token itself becomes a 401. Reporting a server-side
+    // failure as an invalid credential would have clients discard their refresh token and
+    // re-authorize, turning a brief outage into a re-consent storm.
+    if (error instanceof OAuthError && error.statusCode < 500) {
+      logger.info('Rejected OAuth access token', {
+        error: error.errorCode,
+        description: error.description,
+      });
+      throw invalidToken;
+    }
+
+    logger.error('OAuth authentication failed for a non-token reason', error);
+    throw error;
+  }
+}
+
 type GenerateEngineTokenParams = {
   projectId: ProjectId;
   queueToken?: string;

packages/server/api/src/app/core/security/authn/access-token-authn-handler.ts

@@ -24,17 +24,22 @@ export class AccessTokenAuthnHandler extends BaseSecurityHandler {
     return Promise.resolve(hasToken || !publicRoute);
   }
 
+  /**
+   * An explicit `Authorization` header wins over the session cookie: a caller presenting a
+   * bearer token states which identity it wants, and preferring an ambient cookie would
+   * authenticate it as somebody else.
+   */
   private getAccessToken(request: FastifyRequest): string | undefined {
-    const cookieToken = request.cookies?.[AccessTokenAuthnHandler.COOKIE_NAME];
-    if (!isNil(cookieToken)) {
-      return cookieToken;
-    }
-
     const header = request.headers[AccessTokenAuthnHandler.HEADER_NAME];
     if (header?.startsWith(AccessTokenAuthnHandler.HEADER_PREFIX)) {
       return header.substring(AccessTokenAuthnHandler.HEADER_PREFIX.length);
     }
 
+    const cookieToken = request.cookies?.[AccessTokenAuthnHandler.COOKIE_NAME];
+    if (!isNil(cookieToken)) {
+      return cookieToken;
+    }
+
     return undefined;
   }
 

packages/server/api/src/app/database/database-connection.ts

@@ -23,6 +23,14 @@ import { FlowEntity } from '../flows/flow/flow.entity';
 import { FolderEntity } from '../flows/folder/folder.entity';
 import { FlowStepTestOutputEntity } from '../flows/step-test-output/flow-step-test-output-entity';
 import { TriggerEventEntity } from '../flows/trigger-events/trigger-event.entity';
+import {
+  OAuthAuthorizationCodeEntity,
+  OAuthClientEntity,
+  OAuthGrantEntity,
+  OAuthPendingAuthorizationEntity,
+  OAuthRefreshTokenEntity,
+  OAuthSigningKeyEntity,
+} from '../oauth/storage/oauth.entity';
 import { OrganizationEntity } from '../organization/organization.entity';
 import { ProjectEntity } from '../project/project-entity';
 import { StoreEntryEntity } from '../store-entry/store-entry-entity';
@@ -60,6 +68,12 @@ function getEntities(): EntitySchema<unknown>[] {
     AiConfigEntity,
     McpConfigEntity,
     FlowStepTestOutputEntity,
+    OAuthSigningKeyEntity,
+    OAuthClientEntity,
+    OAuthGrantEntity,
+    OAuthPendingAuthorizationEntity,
+    OAuthAuthorizationCodeEntity,
+    OAuthRefreshTokenEntity,
   ];
 
   return entities;

packages/server/api/src/app/database/migrations/1786355547856-CreateOAuthTables.ts

@@ -0,0 +1,171 @@
+import { logger } from '@openops/server-shared';
+import { MigrationInterface, QueryRunner } from 'typeorm';
+
+export class CreateOAuthTables1786355547856 implements MigrationInterface {
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    logger.info('CreateOAuthTables1786355547856: starting');
+
+    await queryRunner.query(`
+      CREATE TABLE "oauth_signing_key" (
+        "id" varchar(21) NOT NULL,
+        "created" timestamp with time zone DEFAULT now() NOT NULL,
+        "updated" timestamp with time zone DEFAULT now() NOT NULL,
+        "privateKeyEncrypted" text NOT NULL,
+        "publicKeyPem" text NOT NULL,
+        "status" varchar(16) NOT NULL,
+        CONSTRAINT "PK_oauth_signing_key" PRIMARY KEY ("id")
+      );
+    `);
+
+    // Guarantees concurrently booting replicas converge on a single active key.
+    await queryRunner.query(`
+      CREATE UNIQUE INDEX "idx_oauth_signing_key_single_active"
+      ON "oauth_signing_key" ("status") WHERE "status" = 'active';
+    `);
+
+    await queryRunner.query(`
+      CREATE TABLE "oauth_client" (
+        "id" varchar(21) NOT NULL,
+        "created" timestamp with time zone DEFAULT now() NOT NULL,
+        "updated" timestamp with time zone DEFAULT now() NOT NULL,
+        "clientName" varchar(128) NOT NULL,
+        "redirectUris" jsonb NOT NULL,
+        "grantTypes" jsonb NOT NULL,
+        "tokenEndpointAuthMethod" varchar(32) NOT NULL,
+        "clientSecretHash" varchar(64),
+        CONSTRAINT "PK_oauth_client" PRIMARY KEY ("id")
+      );
+    `);
+
+    await queryRunner.query(`
+      CREATE TABLE "oauth_grant" (
+        "id" varchar(21) NOT NULL,
+        "created" timestamp with time zone DEFAULT now() NOT NULL,
+        "updated" timestamp with time zone DEFAULT now() NOT NULL,
+        "clientId" varchar(21) NOT NULL,
+        "userId" varchar(21) NOT NULL,
+        "resourceId" varchar(32) NOT NULL,
+        "status" varchar(16) NOT NULL,
+        "lastUsedAt" timestamp with time zone,
+        "revokedAt" timestamp with time zone,
+        CONSTRAINT "PK_oauth_grant" PRIMARY KEY ("id"),
+        CONSTRAINT "fk_oauth_grant_client" FOREIGN KEY ("clientId")
+          REFERENCES "oauth_client" ("id") ON DELETE CASCADE,
+        CONSTRAINT "fk_oauth_grant_user" FOREIGN KEY ("userId")
+          REFERENCES "user" ("id") ON DELETE CASCADE
+      );
+    `);
+
+    // Not unique: a user may hold several connections for the same client.
+    await queryRunner.query(`
+      CREATE INDEX "idx_oauth_grant_client_id_user_id"
+      ON "oauth_grant" ("clientId", "userId");
+    `);
+
+    await queryRunner.query(`
+      CREATE INDEX "idx_oauth_grant_user_id" ON "oauth_grant" ("userId");
+    `);
+
+    await queryRunner.query(`
+      CREATE TABLE "oauth_pending_authorization" (
+        "id" varchar(21) NOT NULL,
+        "created" timestamp with time zone DEFAULT now() NOT NULL,
+        "updated" timestamp with time zone DEFAULT now() NOT NULL,
+        "clientId" varchar(21) NOT NULL,
+        "redirectUri" varchar(512) NOT NULL,
+        "codeChallenge" varchar(43) NOT NULL,
+        "resource" varchar(512) NOT NULL,
+        "scope" varchar(128) NOT NULL,
+        "state" text,
+        "expiresAt" timestamp with time zone NOT NULL,
+        "consumedAt" timestamp with time zone,
+        CONSTRAINT "PK_oauth_pending_authorization" PRIMARY KEY ("id"),
+        CONSTRAINT "fk_oauth_pending_authorization_client" FOREIGN KEY ("clientId")
+          REFERENCES "oauth_client" ("id") ON DELETE CASCADE
+      );
+    `);
+
+    await queryRunner.query(`
+      CREATE INDEX "idx_oauth_pending_authorization_expires_at"
+      ON "oauth_pending_authorization" ("expiresAt");
+    `);
+
+    await queryRunner.query(`
+      CREATE TABLE "oauth_authorization_code" (
+        "id" varchar(21) NOT NULL,
+        "created" timestamp with time zone DEFAULT now() NOT NULL,
+        "updated" timestamp with time zone DEFAULT now() NOT NULL,
+        "codeHash" varchar(64) NOT NULL,
+        "clientId" varchar(21) NOT NULL,
+        "userId" varchar(21) NOT NULL,
+        "redirectUri" varchar(512) NOT NULL,
+        "codeChallenge" varchar(43) NOT NULL,
+        "resource" varchar(512) NOT NULL,
+        "scope" varchar(128) NOT NULL,
+        "expiresAt" timestamp with time zone NOT NULL,
+        "consumedAt" timestamp with time zone,
+        CONSTRAINT "PK_oauth_authorization_code" PRIMARY KEY ("id"),
+        CONSTRAINT "fk_oauth_authorization_code_client" FOREIGN KEY ("clientId")
+          REFERENCES "oauth_client" ("id") ON DELETE CASCADE
+      );
+    `);
+
+    await queryRunner.query(`
+      CREATE UNIQUE INDEX "idx_oauth_authorization_code_code_hash"
+      ON "oauth_authorization_code" ("codeHash");
+    `);
+
+    await queryRunner.query(`
+      CREATE INDEX "idx_oauth_authorization_code_expires_at"
+      ON "oauth_authorization_code" ("expiresAt");
+    `);
+
+    await queryRunner.query(`
+      CREATE TABLE "oauth_refresh_token" (
+        "id" varchar(21) NOT NULL,
+        "created" timestamp with time zone DEFAULT now() NOT NULL,
+        "updated" timestamp with time zone DEFAULT now() NOT NULL,
+        "tokenHash" varchar(64) NOT NULL,
+        "grantId" varchar(21) NOT NULL,
+        "familyId" varchar(21) NOT NULL,
+        "clientId" varchar(21) NOT NULL,
+        "resource" varchar(512) NOT NULL,
+        "scope" varchar(128) NOT NULL,
+        "projectId" varchar(21) NOT NULL,
+        "expiresAt" timestamp with time zone NOT NULL,
+        "revokedAt" timestamp with time zone,
+        CONSTRAINT "PK_oauth_refresh_token" PRIMARY KEY ("id"),
+        CONSTRAINT "fk_oauth_refresh_token_grant" FOREIGN KEY ("grantId")
+          REFERENCES "oauth_grant" ("id") ON DELETE CASCADE,
+        CONSTRAINT "fk_oauth_refresh_token_client" FOREIGN KEY ("clientId")
+          REFERENCES "oauth_client" ("id") ON DELETE CASCADE
+      );
+    `);
+
+    await queryRunner.query(`
+      CREATE UNIQUE INDEX "idx_oauth_refresh_token_token_hash"
+      ON "oauth_refresh_token" ("tokenHash");
+    `);
+
+    await queryRunner.query(`
+      CREATE INDEX "idx_oauth_refresh_token_grant_id"
+      ON "oauth_refresh_token" ("grantId");
+    `);
+
+    await queryRunner.query(`
+      CREATE INDEX "idx_oauth_refresh_token_family_id"
+      ON "oauth_refresh_token" ("familyId");
+    `);
+
+    await queryRunner.query(`
+      CREATE INDEX "idx_oauth_refresh_token_expires_at"
+      ON "oauth_refresh_token" ("expiresAt");
+    `);
+
+    logger.info('CreateOAuthTables1786355547856: completed');
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    throw new Error('Rollback not implemented');
+  }
+}

packages/server/api/src/app/database/postgres-connection.ts

@@ -39,6 +39,7 @@ import { AddBenchmarkAndBenchmarkFlowTables1770297289194 } from './migrations/17
 import { DropLastRunIdFromBenchmark1772449919844 } from './migrations/1772449919844-DropLastRunIdFromBenchmark';
 import { AddIsCleanupToBenchmarkFlow1773046640936 } from './migrations/1773046640936-AddIsCleanupToBenchmarkFlow';
 import { FixFolderUniqueConstraint1776097737024 } from './migrations/1776097737024-FixFolderUniqueConstraint';
+import { CreateOAuthTables1786355547856 } from './migrations/1786355547856-CreateOAuthTables';
 
 const getSslConfig = (): boolean | TlsOptions => {
   const useSsl = system.get(AppSystemProp.POSTGRES_USE_SSL);
@@ -90,6 +91,7 @@ const getMigrations = (): (new () => MigrationInterface)[] => {
     DropLastRunIdFromBenchmark1772449919844,
     AddIsCleanupToBenchmarkFlow1773046640936,
     FixFolderUniqueConstraint1776097737024,
+    CreateOAuthTables1786355547856,
   ];
 };
 

packages/server/api/src/app/flags/flag.service.ts

@@ -10,6 +10,7 @@ import { Flag, FlagId } from '@openops/shared';
 import axios from 'axios';
 import { webhookUtils } from 'server-worker';
 import { repoFactory } from '../core/db/repo-factory';
+import { oauthConfig } from '../oauth/config/oauth-config';
 import { devFlagsService } from './dev-flags.service';
 import { FlagEntity } from './flag.entity';
 import { defaultTheme } from './theme';
@@ -277,6 +278,12 @@ export const flagService = {
         created,
         updated,
       },
+      {
+        id: FlagId.CONNECTED_APPS_ENABLED,
+        value: oauthConfig.isEnabled(),
+        created,
+        updated,
+      },
       {
         id: FlagId.THIRD_PARTY_AUTH_PROVIDER_REDIRECT_URL,
         value: await this.getBackendRedirectUrl(),