feat: complete the two deferred review items (magic-link audience, OTP expiry filter)

Both were noted as follow-up in the prior commit; they land here with the
full `./gradlew check` gate green (Spotless + tests incl. Testcontainers +
JaCoCo).

magic-link: dedicated audience (token-confusion defense)
- MagicLinkService.DEFAULT_AUDIENCE ("pkauth:magic-link") + a new
  Dependencies.ofDedicatedAudience(keyset, issuerName, ...) factory that
  builds a magic-link-scoped JWT issuer + validator (no-op AccessTokenStore,
  so magic-link jtis never enter the access-token store). Because magic-link
  tokens now carry the magic-link audience, the host's resource-server
  validator (application audience) rejects them — a magic-link token can no
  longer be replayed as an API bearer/access token.
- All three adapters (Spring autoconfigure, Dropwizard Dagger, Micronaut
  factory) wire the magic-link service via ofDedicatedAudience using the
  shared keyset + the resource-server issuer name.
- Adds a test proving a magic-link token is rejected by a resource-server
  validator yet still consumed by its own service.

otp: in-store expiry filtering
- OtpRepository.findLatestActive gains an `Instant now` parameter so each
  backend can filter `expiresAt > now` against the HOST clock (not the DB
  wall clock, which would be a second, uncontrollable clock source and broke
  fixed-clock tests). OtpService passes clockProvider.now(); the existing
  service-level expiry re-check stays as defense-in-depth.
- JDBI (`AND expires_at > :now`), DynamoDB (stream filter on parsed
  expiresAt), and the in-memory testkit all filter expiry now.
- OtpServiceTest.expiredOtpIsRejected now drives the service-level re-check
  via a non-filtering repository decorator, preserving that branch's coverage.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: wolpert

pk-auth-core/src/main/java/com/codeheadsystems/pkauth/spi/OtpRepository.java

@@ -57,10 +57,16 @@ record StoredOtp(
   void save(StoredOtp otp);
 
   /**
-   * Returns the most recently issued, non-consumed, non-expired OTP for the given user + phone, if
-   * any. Used by {@code OtpService.verify} as the candidate for matching.
+   * Returns the most recently issued, non-consumed, and non-expired ({@code expiresAt > now}) OTP
+   * for the given user + phone, if any. Used by {@code OtpService.verify} as the candidate for
+   * matching.
+   *
+   * @param userHandle owning user
+   * @param phoneE164 destination phone in E.164 format
+   * @param now the caller's current instant (from the host ClockProvider)
+   * @since 2.1.0
    */
-  Optional<StoredOtp> findLatestActive(UserHandle userHandle, String phoneE164);
+  Optional<StoredOtp> findLatestActive(UserHandle userHandle, String phoneE164, Instant now);
 
   /**
    * Atomically increments the attempts counter for the supplied OTP id and returns the new count.

pk-auth-core/src/test/java/com/codeheadsystems/pkauth/lifecycle/UserDeletionServiceTest.java

@@ -172,7 +172,8 @@ private static final class StubOtpRepository implements OtpRepository {
     public void save(StoredOtp otp) {}
 
     @Override
-    public Optional<StoredOtp> findLatestActive(UserHandle userHandle, String phoneE164) {
+    public Optional<StoredOtp> findLatestActive(
+        UserHandle userHandle, String phoneE164, Instant now) {
       return Optional.empty();
     }
 

pk-auth-dropwizard/src/main/java/com/codeheadsystems/pkauth/dropwizard/dagger/AltFlowsModule.java

@@ -8,8 +8,8 @@
 import com.codeheadsystems.pkauth.backupcodes.BackupCodeService;
 import com.codeheadsystems.pkauth.dropwizard.admin.PkAuthAdminResource;
 import com.codeheadsystems.pkauth.dropwizard.config.PkAuthConfig;
-import com.codeheadsystems.pkauth.jwt.PkAuthJwtIssuer;
-import com.codeheadsystems.pkauth.jwt.PkAuthJwtValidator;
+import com.codeheadsystems.pkauth.jwt.JwtConfig;
+import com.codeheadsystems.pkauth.jwt.JwtKeyset;
 import com.codeheadsystems.pkauth.lifecycle.BackupCodeRepositoryDeletionListener;
 import com.codeheadsystems.pkauth.lifecycle.OtpRepositoryDeletionListener;
 import com.codeheadsystems.pkauth.lifecycle.UserDeletionListener;
@@ -156,8 +156,8 @@ BackupCodeService provideBackupCodeService(BackupCodeRepository repo, ClockProvi
   @Provides
   @Singleton
   MagicLinkService provideMagicLinkService(
-      PkAuthJwtIssuer issuer,
-      PkAuthJwtValidator validator,
+      JwtConfig jwtConfig,
+      JwtKeyset keyset,
       EmailSender emailSender,
       UserLookup userLookup,
       ClockProvider clock,
@@ -168,8 +168,11 @@ MagicLinkService provideMagicLinkService(
           "pkAuth.magicLink configuration block is required when alt-flow auto-wiring is enabled"
               + " (set magicLink.baseUrl).");
     }
+    // Dedicated magic-link issuer/validator scoped to MagicLinkService.DEFAULT_AUDIENCE so the
+    // resource-server validator (application audience) rejects magic-link tokens as bearer tokens.
     return MagicLinkService.create(
-        MagicLinkService.Dependencies.of(issuer, validator, emailSender, userLookup, clock),
+        MagicLinkService.Dependencies.ofDedicatedAudience(
+            keyset, jwtConfig.issuer(), emailSender, userLookup, clock),
         ml.baseUrl());
   }
 

pk-auth-magic-link/src/main/java/com/codeheadsystems/pkauth/magiclink/MagicLinkService.java

@@ -4,6 +4,8 @@
 import com.codeheadsystems.pkauth.api.UserHandle;
 import com.codeheadsystems.pkauth.jwt.AuthMethod;
 import com.codeheadsystems.pkauth.jwt.JwtClaims;
+import com.codeheadsystems.pkauth.jwt.JwtConfig;
+import com.codeheadsystems.pkauth.jwt.JwtKeyset;
 import com.codeheadsystems.pkauth.jwt.JwtVerificationResult;
 import com.codeheadsystems.pkauth.jwt.PkAuthJwtIssuer;
 import com.codeheadsystems.pkauth.jwt.PkAuthJwtValidator;
@@ -80,6 +82,18 @@ public final class MagicLinkService {
   /** Default rate-limit window. */
   public static final Duration DEFAULT_RATE_WINDOW = Duration.ofHours(1);
 
+  /**
+   * Dedicated audience for magic-link JWTs. Magic-link tokens are minted with this audience (not
+   * the application's resource-server audience) so that the host's ordinary {@link
+   * PkAuthJwtValidator} — which only accepts the application audience — rejects them. This is what
+   * prevents a magic-link token (which sits in an email inbox / proxy log) from being replayed as
+   * an API bearer/access token. Wire the service via {@link Dependencies#ofDedicatedAudience} so
+   * the magic-link issuer and validator are both scoped to this audience.
+   *
+   * @since 2.1.0
+   */
+  public static final String DEFAULT_AUDIENCE = "pkauth:magic-link";
+
   /** Result of a send attempt. */
   public sealed interface SendResult {
     /** Email was dispatched. */
@@ -432,6 +446,52 @@ public static Dependencies of(
           new InMemoryConsumedJtiStore(DEFAULT_CONSUMED_JTI_TTL),
           new DefaultMagicLinkFormatter());
     }
+
+    /**
+     * Builds {@link Dependencies} with a <strong>dedicated</strong> magic-link JWT issuer and
+     * validator, both scoped to {@link MagicLinkService#DEFAULT_AUDIENCE} and derived from the
+     * host's {@code keyset} and {@code issuerName} (pass the resource-server issuer so the {@code
+     * iss} claim is consistent). This is the recommended wiring: because magic-link tokens carry
+     * the magic-link audience rather than the application audience, the host's resource-server
+     * {@link PkAuthJwtValidator} rejects them, so a magic-link token cannot be replayed as an API
+     * bearer/access token (token-confusion defense). The dedicated issuer also uses a no-op
+     * access-token store, so magic-link jtis never pollute the host's {@code AccessTokenStore}.
+     *
+     * <p>Prefer this over {@link #of} for production wiring. Token lifetime is controlled by {@link
+     * MagicLinkService} (the {@link Config#tokenTtl()}), not the issuer's access-token TTL, so the
+     * audience config's TTL policy is irrelevant here.
+     *
+     * @param keyset the host's signing keyset (shared with the resource-server issuer)
+     * @param issuerName the {@code iss} claim value (the resource-server issuer name)
+     * @param emailSender the email transport
+     * @param userLookup the user lookup SPI
+     * @param clockProvider the clock
+     * @return dependencies wired to a magic-link-scoped issuer + validator
+     * @since 2.1.0
+     */
+    public static Dependencies ofDedicatedAudience(
+        JwtKeyset keyset,
+        String issuerName,
+        EmailSender emailSender,
+        UserLookup userLookup,
+        ClockProvider clockProvider) {
+      Objects.requireNonNull(keyset, "keyset");
+      Objects.requireNonNull(issuerName, "issuerName");
+      JwtConfig audienceConfig = JwtConfig.defaults(issuerName, DEFAULT_AUDIENCE);
+      // No-op AccessTokenStore on purpose: magic-link jtis must not be recorded as live access
+      // tokens. Single-use is enforced separately via the ConsumedJtiStore.
+      PkAuthJwtIssuer dedicatedIssuer = new PkAuthJwtIssuer(audienceConfig, keyset, clockProvider);
+      PkAuthJwtValidator dedicatedValidator =
+          new PkAuthJwtValidator(audienceConfig, keyset, clockProvider);
+      return new Dependencies(
+          dedicatedIssuer,
+          dedicatedValidator,
+          emailSender,
+          userLookup,
+          clockProvider,
+          new InMemoryConsumedJtiStore(DEFAULT_CONSUMED_JTI_TTL),
+          new DefaultMagicLinkFormatter());
+    }
   }
 
   /**

pk-auth-magic-link/src/test/java/com/codeheadsystems/pkauth/magiclink/MagicLinkServiceTest.java

@@ -7,6 +7,7 @@
 import com.codeheadsystems.pkauth.api.UserHandle;
 import com.codeheadsystems.pkauth.jwt.JwtConfig;
 import com.codeheadsystems.pkauth.jwt.JwtKeyset;
+import com.codeheadsystems.pkauth.jwt.JwtVerificationResult;
 import com.codeheadsystems.pkauth.jwt.PkAuthJwtIssuer;
 import com.codeheadsystems.pkauth.jwt.PkAuthJwtValidator;
 import com.codeheadsystems.pkauth.spi.ClockProvider;
@@ -140,6 +141,35 @@ void finishVerificationRejectsCrossPurposeTokenWithoutConsumingIt() {
             s -> assertThat(s.userHandle()).isEqualTo(user));
   }
 
+  @Test
+  void magicLinkTokenIsRejectedByResourceServerValidatorButConsumedByItsOwnService() {
+    byte[] secret = new byte[32];
+    new SecureRandom().nextBytes(secret);
+    JwtKeyset keyset = JwtKeyset.hs256(secret);
+    ClockProvider clock = ClockProvider.fromClock(Clock.fixed(NOW, ZoneOffset.UTC));
+    users.register("alice", "Alice", "alice@example.com");
+
+    // Service wired with a dedicated magic-link audience.
+    MagicLinkService dedicated =
+        MagicLinkService.create(
+            MagicLinkService.Dependencies.ofDedicatedAudience(keyset, ISSUER, emails, users, clock),
+            BASE_URL);
+    String token =
+        ((MagicLinkService.SendResult.Sent) dedicated.startLogin("alice", "alice@example.com"))
+            .tokenJti();
+
+    // A resource-server validator scoped to the application audience MUST reject the magic-link
+    // token (wrong audience) — it cannot be replayed as an API bearer/access token.
+    PkAuthJwtValidator resourceServer =
+        new PkAuthJwtValidator(JwtConfig.defaults(ISSUER, AUDIENCE), keyset, clock);
+    assertThat(resourceServer.validate(token))
+        .isInstanceOf(JwtVerificationResult.WrongAudience.class);
+
+    // The magic-link service still validates and consumes its own token.
+    assertThat(dedicated.finishVerification(token, MagicLinkService.PURPOSE_LOGIN))
+        .isInstanceOf(MagicLinkService.ConsumeResult.Success.class);
+  }
+
   @Test
   void loginUnknownUserReturnsSentToPreventEnumeration() {
     // Privacy invariant: startLogin must return Sent even when the username does not exist,

pk-auth-micronaut/src/main/java/com/codeheadsystems/pkauth/micronaut/PkAuthFactory.java

@@ -249,13 +249,16 @@ SmsSender smsSender() {
 
   @Singleton
   MagicLinkService magicLinkService(
-      PkAuthJwtIssuer issuer,
-      PkAuthJwtValidator validator,
+      JwtConfig jwtConfig,
+      JwtKeyset keyset,
       EmailSender emailSender,
       UserLookup userLookup,
       ClockProvider clock) {
+    // Dedicated magic-link issuer/validator scoped to MagicLinkService.DEFAULT_AUDIENCE so the
+    // resource-server validator (application audience) rejects magic-link tokens as bearer tokens.
     return MagicLinkService.create(
-        MagicLinkService.Dependencies.of(issuer, validator, emailSender, userLookup, clock),
+        MagicLinkService.Dependencies.ofDedicatedAudience(
+            keyset, jwtConfig.issuer(), emailSender, userLookup, clock),
         "http://localhost:8080/auth/magic");
   }
 

pk-auth-otp/src/main/java/com/codeheadsystems/pkauth/otp/OtpService.java

@@ -181,7 +181,8 @@ public VerifyResult finishVerification(UserHandle user, String phoneE164, String
     Objects.requireNonNull(phoneE164, "phoneE164");
     Objects.requireNonNull(candidate, "candidate");
 
-    Optional<StoredOtp> activeOpt = repository.findLatestActive(user, phoneE164);
+    Instant now = clockProvider.now();
+    Optional<StoredOtp> activeOpt = repository.findLatestActive(user, phoneE164, now);
     if (activeOpt.isEmpty()) {
       // Run a throwaway HMAC so the no-active-OTP branch takes approximately the same wall-clock
       // time as the matching/mismatching branches below — the HMAC dominates the cost, so the
@@ -192,7 +193,6 @@ public VerifyResult finishVerification(UserHandle user, String phoneE164, String
       return new VerifyResult.NoActiveOtp();
     }
     StoredOtp active = activeOpt.get();
-    Instant now = clockProvider.now();
     if (now.isAfter(active.expiresAt())) {
       return new VerifyResult.Expired();
     }

pk-auth-otp/src/test/java/com/codeheadsystems/pkauth/otp/OtpServiceTest.java

@@ -6,6 +6,7 @@
 
 import com.codeheadsystems.pkauth.api.UserHandle;
 import com.codeheadsystems.pkauth.spi.ClockProvider;
+import com.codeheadsystems.pkauth.spi.OtpRepository;
 import com.codeheadsystems.pkauth.testkit.InMemoryOtpRepository;
 import java.security.SecureRandom;
 import java.time.Clock;
@@ -14,6 +15,8 @@
 import java.time.ZoneOffset;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Optional;
+import java.util.OptionalInt;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
@@ -125,10 +128,14 @@ void sendRateLimits() {
   void expiredOtpIsRejected() {
     service.startVerification(USER, PHONE);
     String code = sms.lastCode();
+    // Wrap the repository so it does NOT filter expiry — this exercises the service's own expiry
+    // re-check (defense-in-depth). A conforming repository filters expired rows itself (verified by
+    // the persistence integration tests); this asserts the service still rejects an expired row if
+    // a host repository hands one back.
     OtpService advanced =
         OtpService.create(
             OtpService.Dependencies.of(
-                repository,
+                new NonFilteringOtpRepository(repository),
                 sms,
                 ClockProvider.fromClock(
                     Clock.fixed(NOW.plus(Duration.ofMinutes(10)), ZoneOffset.UTC))),
@@ -143,6 +150,50 @@ void expiredOtpIsRejected() {
         .isInstanceOf(OtpService.VerifyResult.Expired.class);
   }
 
+  /**
+   * {@link OtpRepository} decorator that ignores the {@code now} expiry filter (delegates with
+   * {@link Instant#MIN}), modelling a host repository that does not filter expired rows in-store —
+   * so the service-level expiry re-check is the thing under test.
+   */
+  private static final class NonFilteringOtpRepository implements OtpRepository {
+    private final OtpRepository delegate;
+
+    NonFilteringOtpRepository(OtpRepository delegate) {
+      this.delegate = delegate;
+    }
+
+    @Override
+    public void save(StoredOtp otp) {
+      delegate.save(otp);
+    }
+
+    @Override
+    public Optional<StoredOtp> findLatestActive(
+        UserHandle userHandle, String phoneE164, Instant now) {
+      return delegate.findLatestActive(userHandle, phoneE164, Instant.MIN);
+    }
+
+    @Override
+    public OptionalInt incrementAttempts(UserHandle userHandle, String otpId) {
+      return delegate.incrementAttempts(userHandle, otpId);
+    }
+
+    @Override
+    public boolean consume(UserHandle userHandle, String otpId) {
+      return delegate.consume(userHandle, otpId);
+    }
+
+    @Override
+    public int countSince(UserHandle userHandle, String phoneE164, Instant since) {
+      return delegate.countSince(userHandle, phoneE164, since);
+    }
+
+    @Override
+    public int deleteByUserHandle(UserHandle userHandle) {
+      return delegate.deleteByUserHandle(userHandle);
+    }
+  }
+
   @Test
   void maskPhoneKeepsCountryPrefixAndLast4() {
     // Normal E.164 numbers: keep '+' + first country digit + '***' + last 4 digits.

pk-auth-persistence-dynamodb/src/main/java/com/codeheadsystems/pkauth/persistence/dynamodb/DynamoDbOtpRepository.java

@@ -49,7 +49,8 @@ public void save(StoredOtp otp) {
   }
 
   @Override
-  public Optional<StoredOtp> findLatestActive(UserHandle userHandle, String phoneE164) {
+  public Optional<StoredOtp> findLatestActive(
+      UserHandle userHandle, String phoneE164, Instant now) {
     return DynamoDbSupport.wrap(
         "otp.findLatestActive",
         () -> {
@@ -65,6 +66,9 @@ public Optional<StoredOtp> findLatestActive(UserHandle userHandle, String phoneE
               .flatMap(page -> page.items().stream())
               .filter(i -> phoneE164.equals(i.getPhoneE164()))
               .filter(i -> !i.isConsumed())
+              // Filter expiry against the caller-supplied instant (the host ClockProvider's "now"),
+              // not the DynamoDB wall clock, so a single authoritative clock governs expiry.
+              .filter(i -> DynamoDbSupport.parseInstant(i.getExpiresAt()).isAfter(now))
               // Compare parsed Instants, not the raw ISO strings: createdAt is stored via
               // Instant.toString(), which is variable-precision (it drops the fractional-seconds
               // field when zero), so "...:00Z" sorts after "...:00.000001Z" lexicographically and

pk-auth-persistence-dynamodb/src/test/java/com/codeheadsystems/pkauth/persistence/dynamodb/DynamoDbAltFlowsIntegrationTest.java

@@ -92,8 +92,8 @@ void otpDeleteByUserHandleRemovesEveryRowForTheUser() {
 
     int removed = otp.deleteByUserHandle(user);
     assertThat(removed).isEqualTo(2);
-    assertThat(otp.findLatestActive(user, "+15551110000")).isEmpty();
-    assertThat(otp.findLatestActive(other, "+15552220000")).isPresent();
+    assertThat(otp.findLatestActive(user, "+15551110000", t0)).isEmpty();
+    assertThat(otp.findLatestActive(other, "+15552220000", t0)).isPresent();
   }
 
   @Test
@@ -115,17 +115,17 @@ void otpRoundTripAndCountSince() {
             t0.plusSeconds(60),
             t0.plusSeconds(360)));
 
-    var active = otp.findLatestActive(user, "+15551234567").orElseThrow();
+    var active = otp.findLatestActive(user, "+15551234567", t0).orElseThrow();
     assertThat(active.otpId()).isEqualTo("o2");
 
     otp.incrementAttempts(user, "o2");
-    var refreshed = otp.findLatestActive(user, "+15551234567").orElseThrow();
+    var refreshed = otp.findLatestActive(user, "+15551234567", t0).orElseThrow();
     assertThat(refreshed.attempts()).isEqualTo(1);
 
     assertThat(otp.countSince(user, "+15551234567", t0.minusSeconds(10))).isEqualTo(2);
 
     otp.consume(user, "o2");
-    assertThat(otp.findLatestActive(user, "+15551234567"))
+    assertThat(otp.findLatestActive(user, "+15551234567", t0))
         .hasValueSatisfying(o -> assertThat(o.otpId()).isEqualTo("o1"));
   }