Backend JWT Verification

npm install jose

import { createRemoteJWKSet, jwtVerify } from 'jose'

const JWKS = createRemoteJWKSet(
  new URL('https://api.grantiva.io/.well-known/jwks.json')
)

async function verifyGrantivaToken(token) {
  const { payload } = await jwtVerify(token, JWKS, {
    issuer: 'grantiva',
    algorithms: ['RS256'],
  })

  const riskScore    = payload.risk_score       // number | null
  const riskCategory = payload.risk_category    // "trusted" | "suspicious" | "blocked"
  const jailbreak    = payload.jailbreak_detected // boolean
  const customClaims = payload.custom_claims    // object | undefined

  // Block high-risk devices
  if (riskCategory === 'blocked') {
    throw new Error('Device blocked due to high risk score')
  }

  return payload
}

// Express middleware example
async function requireAttestedDevice(req, res, next) {
  const token = req.headers.authorization?.replace('Bearer ', '')
  if (!token) return res.status(401).json({ error: 'Missing token' })

  try {
    req.deviceClaims = await verifyGrantivaToken(token)
    next()
  } catch (err) {
    res.status(401).json({ error: 'Invalid or expired device token' })
  }
}

pip install PyJWT[crypto] requests

import requests
import jwt
from jwt import PyJWKClient

JWKS_URL = "https://api.grantiva.io/.well-known/jwks.json"

# Reuse the client across requests — it caches the public key
jwks_client = PyJWKClient(JWKS_URL)

def verify_grantiva_token(token: str) -> dict:
    signing_key = jwks_client.get_signing_key_from_jwt(token)

    payload = jwt.decode(
        token,
        signing_key.key,
        algorithms=["RS256"],
        issuer="grantiva",
    )

    risk_score    = payload.get("risk_score")       # int | None
    risk_category = payload.get("risk_category")    # "trusted" | "suspicious" | "blocked"
    jailbreak     = payload.get("jailbreak_detected", False)
    custom_claims = payload.get("custom_claims", {})

    # Block high-risk devices
    if risk_category == "blocked":
        raise ValueError(f"Device blocked: risk_score={risk_score}")

    return payload

# FastAPI / Flask decorator example
from functools import wraps
from flask import request, jsonify, g

def require_attested_device(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        token = request.headers.get("Authorization", "").removeprefix("Bearer ")
        if not token:
            return jsonify({"error": "Missing token"}), 401
        try:
            g.device_claims = verify_grantiva_token(token)
        except Exception as e:
            return jsonify({"error": str(e)}), 401
        return f(*args, **kwargs)
    return decorated

// Package.swift
.package(url: "https://github.com/vapor/jwt.git", from: "5.0.0"),
// target dependency: .product(name: "JWT", package: "jwt")

import Vapor
import JWT

// In configure.swift — fetch and cache Grantiva's public key at startup
public func configure(_ app: Application) async throws {
    let jwksURL = URI(string: "https://api.grantiva.io/.well-known/jwks.json")
    let response = try await app.client.get(jwksURL)
    let jwks = try response.content.decode(JWKS.self)
    try await app.jwt.keys.use(jwks: jwks)
}

// Payload struct matching Grantiva JWT claims
struct GrantivaPayload: JWTPayload {
    enum CodingKeys: String, CodingKey {
        case subject        = "sub"
        case expiration     = "exp"
        case issuer         = "iss"
        case deviceId       = "device_id"
        case riskScore      = "risk_score"
        case riskCategory   = "risk_category"
        case jailbreak      = "jailbreak_detected"
        case bundleId       = "bundle_id"
        case teamId         = "team_id"
        case customClaims   = "custom_claims"
    }

    let subject: SubjectClaim
    let expiration: ExpirationClaim
    let issuer: IssuerClaim
    let deviceId: String
    let riskScore: Int?          // nil on Free tier
    let riskCategory: String     // "trusted" | "suspicious" | "blocked"
    let jailbreak: Bool
    let bundleId: String
    let teamId: String
    let customClaims: [String: String]?

    func verify(using algorithm: some JWTAlgorithm) async throws {
        try expiration.verifyNotExpired()
        guard issuer.value == "grantiva" else {
            throw JWTError.claimVerificationFailure(failedClaim: issuer, reason: "unexpected issuer")
        }
    }
}

// Middleware — verify token on every authenticated request
struct GrantivaDeviceMiddleware: AsyncMiddleware {
    func respond(to request: Request, chainingTo next: any AsyncResponder) async throws -> Response {
        guard let token = request.headers.bearerAuthorization?.token else {
            throw Abort(.unauthorized, reason: "Missing Grantiva device token")
        }

        let payload = try await request.jwt.verify(token, as: GrantivaPayload.self)

        if payload.riskCategory == "blocked" {
            throw Abort(.forbidden, reason: "Device blocked (risk score: \(payload.riskScore ?? -1))")
        }

        request.storage[GrantivaPayloadKey.self] = payload
        return try await next.respond(to: request)
    }
}

private struct GrantivaPayloadKey: StorageKey {
    typealias Value = GrantivaPayload
}

// Usage in routes
app.grouped(GrantivaDeviceMiddleware()).get("protected") { req async throws -> String in
    let device = req.storage[GrantivaPayloadKey.self]!
    return "Hello attested device \(device.deviceId)"
}

go get github.com/golang-jwt/jwt/v5
go get github.com/MicahParks/keyfunc/v3

package main

import (
    "context"
    "errors"
    "fmt"
    "net/http"
    "strings"

    "github.com/MicahParks/keyfunc/v3"
    "github.com/golang-jwt/jwt/v5"
)

const jwksURL = "https://api.grantiva.io/.well-known/jwks.json"

type GrantivaClaims struct {
    DeviceID          string         `json:"device_id"`
    RiskScore         *int           `json:"risk_score"`        // nil on Free tier
    RiskCategory      string         `json:"risk_category"`
    DeviceIntegrity   string         `json:"device_integrity"`
    JailbreakDetected bool           `json:"jailbreak_detected"`
    AttestationCount  int            `json:"attestation_count"`
    CustomClaims      map[string]any `json:"custom_claims"`
    jwt.RegisteredClaims
}

// Initialize once at startup; keyfunc handles key rotation automatically
var jwks keyfunc.Keyfunc

func init() {
    var err error
    jwks, err = keyfunc.NewDefault([]string{jwksURL})
    if err != nil {
        panic(fmt.Sprintf("failed to fetch JWKS: %v", err))
    }
}

func VerifyGrantivaToken(tokenStr string) (*GrantivaClaims, error) {
    claims := &GrantivaClaims{}

    token, err := jwt.ParseWithClaims(tokenStr, claims, jwks.Keyfunc,
        jwt.WithIssuedAt(),
        jwt.WithIssuer("grantiva"),
        jwt.WithValidMethods([]string{"RS256"}),
    )
    if err != nil || !token.Valid {
        return nil, fmt.Errorf("invalid token: %w", err)
    }

    // Block high-risk devices
    if claims.RiskCategory == "blocked" {
        return nil, errors.New("device blocked due to high risk score")
    }

    return claims, nil
}

// HTTP middleware example
func RequireAttestedDevice(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        auth := r.Header.Get("Authorization")
        token := strings.TrimPrefix(auth, "Bearer ")
        if token == "" {
            http.Error(w, "missing token", http.StatusUnauthorized)
            return
        }

        claims, err := VerifyGrantivaToken(token)
        if err != nil {
            http.Error(w, err.Error(), http.StatusUnauthorized)
            return
        }

        ctx := context.WithValue(r.Context(), "device_claims", claims)
        next.ServeHTTP(w, r.WithContext(ctx))
    })
}

// Node.js
const userTier = payload.custom_claims?.user_tier   // "premium"
const region   = payload.custom_claims?.region      // "us-east"

# Python
user_tier = payload.get("custom_claims", {}).get("user_tier")

// Go
userTier, _ := claims.CustomClaims["user_tier"].(string)

// Node.js
const sub = payload.custom_claims?.subscription

if (sub?.is_active && sub.environment === "production") {
  grantAccess(sub.tier)     // your tier strings, e.g. "plus"
} else {
  grantAccess("free")       // absent claim ⇒ free
}

# Python
sub = payload.get("custom_claims", {}).get("subscription")
is_premium = bool(sub and sub["is_active"] and sub["environment"] == "production")

// Allow, warn, or block based on risk category
switch (payload.risk_category) {
  case 'trusted':     // riskScore 0–20
    break             // full access
  case 'suspicious':  // riskScore 21–75
    logSuspiciousDevice(payload.device_id)
    break             // allow but log
  case 'blocked':     // riskScore 76–100
    return res.status(403).json({ error: 'Device blocked' })
}