Add queue stuff

backend/src/db/migrations/20251029234547_add-pki-acme.ts

@@ -107,7 +107,7 @@ export async function up(knex: Knex): Promise<void> {
       t.string("status").notNullable(); // pending, valid, invalid, deactivated, expired, revoked
 
       // Token used to validate the authorization through ACME challenge
-      t.timestamp("token").nullable();
+      t.string("token").nullable();
 
       // Identifier type and value
       t.string("identifierType").notNullable(); // dns

backend/src/db/schemas/pki-acme-auths.ts

@@ -11,7 +11,7 @@ export const PkiAcmeAuthsSchema = z.object({
   id: z.string().uuid(),
   accountId: z.string().uuid(),
   status: z.string(),
-  token: z.date().nullable().optional(),
+  token: z.string().nullable().optional(),
   identifierType: z.string(),
   identifierValue: z.string(),
   expiresAt: z.date(),

backend/src/ee/services/pki-acme/pki-acme-challenge-dal.ts

@@ -9,19 +9,11 @@ export type TPkiAcmeChallengeDALFactory = ReturnType<typeof pkiAcmeChallengeDALF
 export const pkiAcmeChallengeDALFactory = (db: TDbClient) => {
   const pkiAcmeChallengeOrm = ormify(db, TableName.PkiAcmeChallenge);
 
-  const findByAccountAuthAndChallengeIdWithToken = async (
-    accountId: string,
-    authId: string,
-    challengeId: string,
-    tx?: Knex
-  ) => {
+  const findByAccountAuthAndChallengeId = async (accountId: string, authId: string, challengeId: string, tx?: Knex) => {
     try {
       const challenge = await (tx || db)(TableName.PkiAcmeChallenge)
         .join(TableName.PkiAcmeAuth, `${TableName.PkiAcmeChallenge}.authId`, `${TableName.PkiAcmeAuth}.id`)
-        .select(
-          selectAllTableCols(TableName.PkiAcmeChallenge),
-          db.ref("token").withSchema(TableName.PkiAcmeAuth).as("token")
-        )
+        .select(selectAllTableCols(TableName.PkiAcmeChallenge))
         .where(`${TableName.PkiAcmeChallenge}.id`, challengeId)
         .where(`${TableName.PkiAcmeChallenge}.authId`, authId)
         .where(`${TableName.PkiAcmeAuth}.accountId`, accountId)
@@ -34,14 +26,11 @@ export const pkiAcmeChallengeDALFactory = (db: TDbClient) => {
       throw new DatabaseError({ error, name: "Find PKI ACME challenge by account id, auth id and challenge id" });
     }
   };
+
   const findByIdForChallengeValidation = async (id: string, tx?: Knex) => {
-    const rows = await (tx || db)(TableName.PkiAcmeChallenge)
-      .join<TPkiAcmeAuths>(TableName.PkiAcmeAuth, `${TableName.PkiAcmeChallenge}.authId`, `${TableName.PkiAcmeAuth}.id`)
-      .join<TPkiAcmeAccounts>(
-        TableName.PkiAcmeAccount,
-        `${TableName.PkiAcmeAuth}.accountId`,
-        `${TableName.PkiAcmeAccount}.id`
-      )
+    const result = await (tx || db)(TableName.PkiAcmeChallenge)
+      .join(TableName.PkiAcmeAuth, `${TableName.PkiAcmeChallenge}.authId`, `${TableName.PkiAcmeAuth}.id`)
+      .join(TableName.PkiAcmeAccount, `${TableName.PkiAcmeAuth}.accountId`, `${TableName.PkiAcmeAccount}.id`)
       .select(
         selectAllTableCols(TableName.PkiAcmeChallenge),
         db.ref("id").withSchema(TableName.PkiAcmeAuth).as("authId"),
@@ -55,45 +44,41 @@ export const pkiAcmeChallengeDALFactory = (db: TDbClient) => {
       )
       // For all challenges, acquire update lock on the auth to avoid race conditions
       .forUpdate(TableName.PkiAcmeAuth)
-      .where(`${TableName.PkiAcmeChallenge}.id`, id);
-
-    if (rows.length === 0) {
+      .where(`${TableName.PkiAcmeChallenge}.id`, id)
+      .first();
+    if (!result) {
       return null;
     }
-    return sqlNestRelationships({
-      data: rows,
-      key: "id",
-      parentMapper: (row) => row,
-      childrenMapper: [
-        {
-          key: "authId",
-          label: "auth" as const,
-          mapper: ({ authId, authToken, authStatus, authIdentifierType, authIdentifierValue, authExpiresAt }) => ({
-            id: authId,
-            token: authToken,
-            status: authStatus,
-            identifierType: authIdentifierType,
-            identifierValue: authIdentifierValue,
-            expiresAt: authExpiresAt
-          }),
-          childrenMapper: [
-            {
-              key: "accountId",
-              label: "account" as const,
-              mapper: ({ accountId, accountPublicKeyThumbprint }) => ({
-                id: accountId,
-                publicKeyThumbprint: accountPublicKeyThumbprint
-              })
-            }
-          ]
+    const {
+      authId,
+      authToken,
+      authStatus,
+      authIdentifierType,
+      authIdentifierValue,
+      authExpiresAt,
+      accountId,
+      accountPublicKeyThumbprint,
+      ...challenge
+    } = result;
+    return {
+      ...challenge,
+      auth: {
+        token: authToken,
+        status: authStatus,
+        identifierType: authIdentifierType,
+        identifierValue: authIdentifierValue,
+        expiresAt: authExpiresAt,
+        account: {
+          id: accountId,
+          publicKeyThumbprint: accountPublicKeyThumbprint
         }
-      ]
-    })?.[0];
+      }
+    };
   };
 
   return {
     ...pkiAcmeChallengeOrm,
-    findByAccountAuthAndChallengeIdWithToken,
+    findByAccountAuthAndChallengeId,
     findByIdForChallengeValidation
   };
 };

backend/src/ee/services/pki-acme/pki-acme-challenge-queue.ts

@@ -0,0 +1,56 @@
+import { Knex } from "knex";
+
+import { QueueJobs, QueueName, TQueueServiceFactory } from "@app/queue";
+
+import { TPkiAcmeAuthDALFactory } from "./pki-acme-auth-dal";
+import { TPkiAcmeChallengeDALFactory } from "./pki-acme-challenge-dal";
+import { TPkiAcmeChallengeServiceFactory } from "./pki-acme-types";
+
+// Define types for job data
+export type TValidateAcmeChallengeResponseDTO = {
+  challengeId: string;
+};
+
+type TChallengeQueueServiceFactoryDep = {
+  queueService: Pick<TQueueServiceFactory, "queuePg" | "startPg">;
+  acmeChallengeDAL: Pick<TPkiAcmeChallengeDALFactory, "transaction" | "findByIdForChallengeValidation" | "updateById">;
+  acmeAuthDAL: Pick<TPkiAcmeAuthDALFactory, "updateById">;
+  acmeChallengeService: TPkiAcmeChallengeServiceFactory;
+};
+
+export type TFolderCommitQueueServiceFactory = ReturnType<typeof challengeQueueServiceFactory>;
+
+export const challengeQueueServiceFactory = ({
+  queueService,
+  acmeChallengeService
+}: TChallengeQueueServiceFactoryDep) => {
+  const scheduleChallengeValidation = async (payload: TValidateAcmeChallengeResponseDTO) => {
+    const { challengeId } = payload;
+    await queueService.queuePg<QueueName.PkiAcmeChallengeValidation>(QueueJobs.ValidateAcmeChallengeResponse, payload, {
+      // TODO: maybe we should retry, but let's keep it simple for now
+    });
+  };
+
+  const validateAcmeChallengeResponse = async (jobData: TValidateAcmeChallengeResponseDTO, tx?: Knex) => {
+    const { challengeId } = jobData;
+    await acmeChallengeService.validateChallengeResponse(challengeId);
+  };
+
+  const init = async () => {
+    await queueService.startPg<QueueName.PkiAcmeChallengeValidation>(
+      QueueJobs.ValidateAcmeChallengeResponse,
+      async ([job]) => {
+        await validateAcmeChallengeResponse(job.data as TValidateAcmeChallengeResponseDTO);
+      },
+      {
+        workerCount: 5,
+        pollingIntervalSeconds: 30
+      }
+    );
+  };
+
+  return {
+    scheduleChallengeValidation,
+    init
+  };
+};

backend/src/ee/services/pki-acme/pki-acme-challenge-service.ts

@@ -1,21 +1,25 @@
+import { Knex } from "knex";
+
+import { getConfig } from "@app/lib/config/env";
 import { BadRequestError, NotFoundError } from "@app/lib/errors";
+import { TPkiAcmeAuthDALFactory } from "./pki-acme-auth-dal";
 import { TPkiAcmeChallengeDALFactory } from "./pki-acme-challenge-dal";
 import { AcmeAuthStatus, AcmeChallengeStatus, AcmeChallengeType } from "./pki-acme-schemas";
 import { TPkiAcmeChallengeServiceFactory } from "./pki-acme-types";
-import { getConfig } from "@app/lib/config/env";
-import { calculateJwkThumbprint } from "jose";
 
 type TPkiAcmeChallengeServiceFactoryDep = {
-  acmeChallengeDAL: Pick<TPkiAcmeChallengeDALFactory, "transaction" | "findByIdForChallengeValidation">;
+  acmeAuthDAL: Pick<TPkiAcmeAuthDALFactory, "updateById">;
+  acmeChallengeDAL: Pick<TPkiAcmeChallengeDALFactory, "transaction" | "findByIdForChallengeValidation" | "updateById">;
 };
 
 export const pkiAcmeChallengeServiceFactory = ({
+  acmeAuthDAL,
   acmeChallengeDAL
 }: TPkiAcmeChallengeServiceFactoryDep): TPkiAcmeChallengeServiceFactory => {
   const appCfg = getConfig();
 
   const validateChallengeResponse = async (challengeId: string): Promise<void> => {
-    return await acmeChallengeDAL.transaction(async (tx) => {
+    return await acmeChallengeDAL.transaction(async (tx: Knex) => {
       const challenge = await acmeChallengeDAL.findByIdForChallengeValidation(challengeId, tx);
       if (!challenge) {
         throw new NotFoundError({ message: "ACME challenge not found" });
@@ -43,13 +47,27 @@ export const pkiAcmeChallengeServiceFactory = ({
         ? `${baseUrl}:${appCfg.ACME_DEVELOPMENT_HTTP01_CHALLENGE_PORT}`
         : baseUrl;
 
-      const challengeUrl = new URL(`/.well-known/acme-challenge/${challenge.token}`, actualBaseUrl);
+      const challengeUrl = new URL(`/.well-known/acme-challenge/${challenge.auth.token}`, actualBaseUrl);
+      // Notice: well, we are in a transaction, ideally we should not hold transaction and perform
+      //         a long running operation for long time. But assuming we are not performing a tons of
+      //         challenge validation at the same time, it should be fine.
+      // TODO: bound it with timeout of the fetch request
       const challengeResponse = await fetch(challengeUrl);
       if (challengeResponse.status !== 200) {
         throw new BadRequestError({ message: "ACME challenge response is not 200" });
       }
       const challengeResponseBody = await challengeResponse.text();
-      const expectedChallengeResponseBody = `${challenge.token}.${challenge.auth.identifierValue}`;
+      const thumbprint = Buffer.from(challenge.auth.account.publicKeyThumbprint, "utf-8").toString("base64url");
+      const expectedChallengeResponseBody = `${challenge.auth.token}.${thumbprint}`;
+      if (challengeResponseBody !== expectedChallengeResponseBody) {
+        throw new BadRequestError({ message: "ACME challenge response is not correct" });
+      }
+      await acmeChallengeDAL.updateById(
+        challengeId,
+        { status: AcmeChallengeStatus.Valid, validatedAt: new Date() },
+        tx
+      );
+      await acmeAuthDAL.updateById(challenge.auth.account.id, { status: AcmeAuthStatus.Valid }, tx);
     });
   };
 

backend/src/ee/services/pki-acme/pki-acme-order-dal.ts

@@ -13,16 +13,8 @@ export const pkiAcmeOrderDALFactory = (db: TDbClient) => {
   const findByAccountAndOrderIdWithAuthorizations = async (accountId: string, orderId: string, tx?: Knex) => {
     try {
       const rows = await (tx || db)(TableName.PkiAcmeOrder)
-        .join<TPkiAcmeOrderAuths>(
-          TableName.PkiAcmeOrderAuth,
-          `${TableName.PkiAcmeOrderAuth}.orderId`,
-          `${TableName.PkiAcmeOrder}.id`
-        )
-        .join<TPkiAcmeAuths>(
-          TableName.PkiAcmeAuth,
-          `${TableName.PkiAcmeOrderAuth}.authId`,
-          `${TableName.PkiAcmeAuth}.id`
-        )
+        .join(TableName.PkiAcmeOrderAuth, `${TableName.PkiAcmeOrderAuth}.orderId`, `${TableName.PkiAcmeOrder}.id`)
+        .join(TableName.PkiAcmeAuth, `${TableName.PkiAcmeOrderAuth}.authId`, `${TableName.PkiAcmeAuth}.id`)
         .select(
           selectAllTableCols(TableName.PkiAcmeOrder),
           db.ref("id").withSchema(TableName.PkiAcmeAuth).as("authId"),

backend/src/ee/services/pki-acme/pki-acme-service.ts

@@ -2,7 +2,7 @@ import { TPkiAcmeAccounts } from "@app/db/schemas/pki-acme-accounts";
 import { TPkiAcmeAuths } from "@app/db/schemas/pki-acme-auths";
 import { getConfig } from "@app/lib/config/env";
 import { crypto } from "@app/lib/crypto/cryptography";
-import { NotFoundError } from "@app/lib/errors";
+import { BadRequestError, NotFoundError } from "@app/lib/errors";
 import { logger } from "@app/lib/logger";
 import { TCertificateProfileDALFactory } from "@app/services/certificate-profile/certificate-profile-dal";
 
@@ -71,7 +71,10 @@ type TPkiAcmeServiceFactoryDep = {
   acmeOrderDAL: Pick<TPkiAcmeOrderDALFactory, "create" | "transaction" | "findByAccountAndOrderIdWithAuthorizations">;
   acmeAuthDAL: Pick<TPkiAcmeAuthDALFactory, "create" | "findByAccountIdAndAuthIdWithChallenges">;
   acmeOrderAuthDAL: Pick<TPkiAcmeOrderAuthDALFactory, "insertMany">;
-  acmeChallengeDAL: Pick<TPkiAcmeChallengeDALFactory, "create" | "findByAccountAuthAndChallengeIdWithToken">;
+  acmeChallengeDAL: Pick<
+    TPkiAcmeChallengeDALFactory,
+    "create" | "transaction" | "updateById" | "findByAccountAuthAndChallengeId" | "findByIdForChallengeValidation"
+  >;
 };
 
 export const pkiAcmeServiceFactory = ({
@@ -575,7 +578,7 @@ export const pkiAcmeServiceFactory = ({
           type: auth.identifierType,
           value: auth.identifierValue
         },
-        challenges: auth.challenges.map((challenge: TPkiAcmeChallenges) => {
+        challenges: auth.challenges.map((challenge) => {
           return {
             type: challenge.type,
             url: buildUrl(profileId, `/authorizations/${authzId}/challenges/${challenge.id}`),
@@ -602,10 +605,42 @@ export const pkiAcmeServiceFactory = ({
     authzId: string;
     challengeId: string;
   }): Promise<TAcmeResponse<TRespondToAcmeChallengeResponse>> => {
-    const challenge = await acmeChallengeDAL.findByAccountAuthAndChallengeIdWithToken(accountId, authzId, challengeId);
-    if (!challenge) {
+    const result = await acmeChallengeDAL.findByAccountAuthAndChallengeId(accountId, authzId, challengeId);
+    if (!result) {
       throw new NotFoundError({ message: "ACME challenge not found" });
     }
+    const challenge = await acmeChallengeDAL.transaction(async (tx) => {
+      const challenge = await acmeChallengeDAL.findByIdForChallengeValidation(challengeId, tx);
+      if (!challenge) {
+        throw new NotFoundError({ message: "ACME challenge not found" });
+      }
+      if (challenge.status !== AcmeChallengeStatus.Pending) {
+        // Ideally this should be an ACME error, but RFC 8555 doesn't say much about corner cases like this...
+        throw new BadRequestError({
+          message: `ACME challenge is ${challenge.status} instead of ${AcmeChallengeStatus.Pending}`
+        });
+      }
+      if (challenge.auth.expiresAt < new Date()) {
+        throw new BadRequestError({ message: "ACME auth has expired" });
+      }
+      if (challenge.auth.status !== AcmeAuthStatus.Pending) {
+        throw new BadRequestError({
+          message: `ACME auth status is ${challenge.auth.status} instead of ${AcmeAuthStatus.Pending}`
+        });
+      }
+      if (!challenge.auth.token) {
+        throw new AcmeServerInternalError({ message: "ACME challenge token is required" });
+      }
+      const updatedChallenge = await acmeChallengeDAL.updateById(
+        challengeId,
+        { status: AcmeChallengeStatus.Pending },
+        tx
+      );
+      return {
+        ...challenge,
+        ...updatedChallenge
+      };
+    });
     // TODO: Implement ACME challenge response
     return {
       status: 200,
@@ -613,7 +648,7 @@ export const pkiAcmeServiceFactory = ({
         type: challenge.type,
         url: buildUrl(profileId, `/authorizations/${authzId}/challenges/${challengeId}`),
         status: challenge.status,
-        token: challenge.token
+        token: challenge.auth.token!
       },
       headers: {
         Location: buildUrl(profileId, `/authorizations/${authzId}/challenges/${challengeId}`),

backend/src/ee/services/pki-acme/pki-acme-types.ts

@@ -175,4 +175,6 @@ export type TPkiAcmeServiceFactory = {
   }) => Promise<TAcmeResponse<TRespondToAcmeChallengeResponse>>;
 };
 
-export type TPkiAcmeChallengeServiceFactory = {};
+export type TPkiAcmeChallengeServiceFactory = {
+  validateChallengeResponse: (challengeId: string) => Promise<void>;
+};

backend/src/queue/queue-service.ts

@@ -42,6 +42,7 @@ import {
 } from "@app/services/secret-sync/secret-sync-types";
 import { CacheType } from "@app/services/super-admin/super-admin-types";
 import { TWebhookPayloads } from "@app/services/webhook/webhook-types";
+import { TValidateAcmeChallengeResponseDTO } from "@app/ee/services/pki-acme/pki-acme-challenge-queue";
 
 export enum QueueName {
   SecretRotation = "secret-rotation",
@@ -79,7 +80,8 @@ export enum QueueName {
   UserNotification = "user-notification",
   HealthAlert = "health-alert",
   CertificateV3AutoRenewal = "certificate-v3-auto-renewal",
-  PamAccountRotation = "pam-account-rotation"
+  PamAccountRotation = "pam-account-rotation",
+  PkiAcmeChallengeValidation = "pki-acme-challenge-validation"
 }
 
 export enum QueueJobs {
@@ -130,7 +132,8 @@ export enum QueueJobs {
   UserNotification = "user-notification-job",
   HealthAlert = "health-alert",
   CertificateV3DailyAutoRenewal = "certificate-v3-daily-auto-renewal",
-  PamAccountRotation = "pam-account-rotation"
+  PamAccountRotation = "pam-account-rotation",
+  ValidateAcmeChallengeResponse = "validate-acme-challenge-response"
 }
 
 export type TQueueJobTypes = {
@@ -369,6 +372,10 @@ export type TQueueJobTypes = {
     name: QueueJobs.PamAccountRotation;
     payload: undefined;
   };
+  [QueueName.PkiAcmeChallengeValidation]: {
+    name: QueueJobs.ValidateAcmeChallengeResponse;
+    payload: TValidateAcmeChallengeResponseDTO;
+  };
 };
 
 const SECRET_SCANNING_JOBS = [