Auth0

npm install auth0 jose @noresponse/node

pip install auth0-python "pyjwt[crypto]" noresponse

go get github.com/auth0/go-auth0 github.com/coreos/go-oidc/v3 github.com/noresponsehq/go/v5

import { createRemoteJWKSet, jwtVerify } from "jose";
import { ManagementClient } from "auth0";
import { NoResponse, fromAuth0 } from "@noresponse/node";

const domain = process.env.AUTH0_DOMAIN!;
const issuer = process.env.AUTH0_ISSUER ?? `https://${domain}/`;
const jwks = createRemoteJWKSet(new URL(`${issuer}.well-known/jwks.json`));
const management = new ManagementClient({
  domain,
  clientId: process.env.AUTH0_M2M_CLIENT_ID!,
  clientSecret: process.env.AUTH0_M2M_CLIENT_SECRET!,
});
const nr = new NoResponse(); // reads NORESPONSE_API_KEY

// The Management API types its times as string or object; they arrive as RFC 3339 strings.
const time = (value: unknown) => (typeof value === "string" ? value : undefined);

export async function assessAccountCreation(idToken: string, ip: string) {
  // 1. Verify the ID token, then read the user and the token's session.
  const { payload: claims } = await jwtVerify(idToken, jwks, { issuer, audience: process.env.AUTH0_CLIENT_ID! });
  const user = await management.users.get(claims.sub!);
  const session = typeof claims.sid === "string" ? await management.sessions.get(claims.sid) : undefined;

  // amr holding "mfa" means a second factor during this login, at auth_time.
  const amr = Array.isArray(claims.amr) ? claims.amr : [];
  const mfa = amr.includes("mfa") && typeof claims.auth_time === "number"
    ? { present: true, methods: ["mfa"], ageSeconds: Math.max(0, Math.floor(Date.now() / 1000) - claims.auth_time) }
    : undefined;

  // 2. The adapter maps the claims; the records supply the rest.
  const authContext = fromAuth0(claims, {
    accountCreatedAt: time(user.created_at),
    sessionCreatedAt: time(session?.created_at),
    phoneVerified: user.phone_verified,
    mfa,
  });

  // 3. NoResponse assesses the account creation.
  return nr.assessments.create({
    accountId: claims.sub!,
    identifiers: { email: user.email, ip },
    flow: "account_creation",
    authContext,
  });
}

import os
import time
from datetime import datetime, timezone
from typing import Any

import jwt
from auth0.management import ManagementClient
from noresponse import Assessment, NoResponse, from_auth0

domain = os.environ["AUTH0_DOMAIN"]
issuer = os.environ.get("AUTH0_ISSUER", f"https://{domain}/")
jwks = jwt.PyJWKClient(f"{issuer}.well-known/jwks.json")
management = ManagementClient(
    domain=domain,
    client_id=os.environ["AUTH0_M2M_CLIENT_ID"],
    client_secret=os.environ["AUTH0_M2M_CLIENT_SECRET"],
)
nr = NoResponse()  # reads NORESPONSE_API_KEY


def rfc3339(value: Any) -> str | None:
    """auth0-python returns Management API times as datetime."""
    if isinstance(value, str):
        return value
    if not isinstance(value, datetime):
        return None
    return value.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def mfa_from(claims: dict[str, Any]) -> dict[str, Any] | None:
    """amr holding "mfa" means a second factor during this login, at auth_time."""
    amr = claims.get("amr")
    auth_time = claims.get("auth_time")
    if not isinstance(amr, list) or "mfa" not in amr or not isinstance(auth_time, int):
        return None
    return {"present": True, "methods": ["mfa"], "age_seconds": max(0, int(time.time()) - auth_time)}


def assess_account_creation(id_token: str, ip: str) -> Assessment:
    # 1. Verify the ID token, then read the user and the token's session.
    key = jwks.get_signing_key_from_jwt(id_token).key
    claims = jwt.decode(id_token, key, algorithms=["RS256"], issuer=issuer, audience=os.environ["AUTH0_CLIENT_ID"])
    user = management.users.get(claims["sub"])
    session = management.sessions.get(claims["sid"]) if "sid" in claims else None

    # 2. The adapter maps the claims; the records supply the rest.
    auth_context = from_auth0(claims, {
        "account_created_at": rfc3339(user.created_at),
        "session_created_at": rfc3339(session.created_at) if session else None,
        "phone_verified": user.phone_verified,
        "mfa": mfa_from(claims),
    })

    # 3. NoResponse assesses the account creation.
    return nr.assessments.create(
        account_id=claims["sub"],
        identifiers={"email": user.email, "ip": ip},
        flow="account_creation",
        auth_context=auth_context,
    )

package accounts

import (
	"context"
	"os"
	"slices"
	"time"

	"github.com/auth0/go-auth0/management"
	"github.com/coreos/go-oidc/v3/oidc"
	noresponse "github.com/noresponsehq/go/v5"
	"github.com/noresponsehq/go/v5/models/components"
)

var (
	nr       = noresponse.NewClient(os.Getenv("NORESPONSE_API_KEY"))
	verifier *oidc.IDTokenVerifier
	auth0API *management.Management
)

func init() {
	ctx := context.Background()
	domain := os.Getenv("AUTH0_DOMAIN")
	issuer := os.Getenv("AUTH0_ISSUER")
	if issuer == "" {
		issuer = "https://" + domain + "/"
	}
	// NewProvider reads the discovery document and caches the signing keys.
	provider, err := oidc.NewProvider(ctx, issuer)
	if err != nil {
		panic(err)
	}
	verifier = provider.Verifier(&oidc.Config{ClientID: os.Getenv("AUTH0_CLIENT_ID")})
	auth0API, err = management.New(domain, management.WithClientCredentials(ctx, os.Getenv("AUTH0_M2M_CLIENT_ID"), os.Getenv("AUTH0_M2M_CLIENT_SECRET")))
	if err != nil {
		panic(err)
	}
}

func AssessAccountCreation(ctx context.Context, idToken, ip string) (*components.Assessment, error) {
	// 1. Verify the ID token, then read the user and the token's session.
	tok, err := verifier.Verify(ctx, idToken)
	if err != nil {
		return nil, err
	}
	var claims map[string]any
	if err := tok.Claims(&claims); err != nil {
		return nil, err
	}
	usr, err := auth0API.User.Read(ctx, tok.Subject)
	if err != nil {
		return nil, err
	}

	// 2. The adapter maps the claims; the records supply the rest.
	overrides := map[string]any{
		"account_created_at": usr.GetCreatedAt().UTC().Format(time.RFC3339),
	}
	if usr.PhoneVerified != nil {
		overrides["phone_verified"] = *usr.PhoneVerified
	}
	if sid, ok := claims["sid"].(string); ok {
		sess, err := auth0API.Session.Read(ctx, sid)
		if err != nil {
			return nil, err
		}
		overrides["session_created_at"] = sess.GetCreatedAt() // go-auth0 keeps it as an RFC 3339 string
	}
	// amr holding "mfa" means a second factor during this login, at auth_time.
	amr, _ := claims["amr"].([]any)
	if authTime, ok := claims["auth_time"].(float64); ok && slices.Contains(amr, any("mfa")) {
		age := max(0, time.Now().Unix()-int64(authTime))
		overrides["mfa"] = map[string]any{"present": true, "methods": []string{"mfa"}, "age_seconds": age}
	}
	authContext, err := noresponse.FromAuth0Claims(claims, overrides)
	if err != nil {
		return nil, err
	}

	// 3. NoResponse assesses the account creation. The second argument is the
	// Idempotency-Key, which the SDK reuses across its own retries.
	res, err := nr.Assessments.Create(ctx, noresponse.NewIdempotencyKey(), components.AssessmentRequest{
		AccountID:   tok.Subject,
		Identifiers: components.AssessmentRequestIdentifiers{Email: usr.Email, IP: noresponse.Pointer(ip)},
		Flow:        components.FlowAccountCreation,
		AuthContext: authContext,
	})
	if err != nil {
		return nil, err
	}
	return &res.Result, nil
}