API Clients

# Run:
#   pip install "PyJWT[crypto]" requests
#   CLIENT_ID=<your-client-id> \
#     KEY_PATH=client.pem \
#     TOKEN_URL=https://auth.<region>.strata.io/oauth2/token \
#     python main.py
import os
import time

import jwt
import requests

CLIENT_ID = os.environ["CLIENT_ID"]
KEY_PATH = os.environ["KEY_PATH"]
TOKEN_URL = os.environ["TOKEN_URL"]

with open(KEY_PATH, "rb") as f:
    private_key = f.read()

now = int(time.time())
claims = {
    "iss": CLIENT_ID,
    "sub": CLIENT_ID,
    "aud": TOKEN_URL,
    "iat": now,
    "exp": now + 300,
}

assertion = jwt.encode(claims, private_key, algorithm="ES256")

resp = requests.post(
    TOKEN_URL,
    data={
        "grant_type": "client_credentials",
        "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
        "client_assertion": assertion,
    },
)
print(resp.status_code, resp.json())

// Run:
//   npm install jose
//   CLIENT_ID=<your-client-id> \
//     KEY_PATH=client.pem \
//     TOKEN_URL=https://auth.<region>.strata.io/oauth2/token \
//     npx tsx main.ts
import { readFile } from "node:fs/promises";
import { SignJWT, importPKCS8 } from "jose";

const CLIENT_ID = process.env.CLIENT_ID!;
const KEY_PATH = process.env.KEY_PATH!;
const TOKEN_URL = process.env.TOKEN_URL!;

interface TokenResponse {
    access_token: string;
    token_type: string;
    expires_in: number;
}

const pem = await readFile(KEY_PATH, "utf8");
const key = await importPKCS8(pem, "ES256");

const assertion = await new SignJWT({})
    .setProtectedHeader({ alg: "ES256", typ: "JWT" })
    .setIssuer(CLIENT_ID)
    .setSubject(CLIENT_ID)
    .setAudience(TOKEN_URL)
    .setIssuedAt()
    .setExpirationTime("5m")
    .sign(key);

const resp = await fetch(TOKEN_URL, {
    method: "POST",
    headers: { "Content-Type": "application/x-www-form-urlencoded" },
    body: new URLSearchParams({
        grant_type: "client_credentials",
        client_assertion_type:
            "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
        client_assertion: assertion,
    }),
});
const data = (await resp.json()) as TokenResponse;
console.log(resp.status, data);

// Run:
//   Cargo.toml dependencies:
//     jsonwebtoken = "9"
//     reqwest = { version = "0.12", features = ["rustls-tls"] }
//     serde = { version = "1", features = ["derive"] }
//     tokio = { version = "1", features = ["full"] }
//   CLIENT_ID=<your-client-id> \
//     KEY_PATH=client.pem \
//     TOKEN_URL=https://auth.<region>.strata.io/oauth2/token \
//     cargo run --release
use std::env;
use std::fs;
use std::time::{SystemTime, UNIX_EPOCH};

use jsonwebtoken::{encode, Algorithm, EncodingKey, Header};
use serde::Serialize;

#[derive(Serialize)]
struct Claims {
    iss: String,
    sub: String,
    aud: String,
    iat: u64,
    exp: u64,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client_id = env::var("CLIENT_ID")?;
    let key_path = env::var("KEY_PATH")?;
    let token_url = env::var("TOKEN_URL")?;

    let pem = fs::read(&key_path)?;
    let key = EncodingKey::from_ec_pem(&pem)?;

    let now = SystemTime::now().duration_since(UNIX_EPOCH)?.as_secs();
    let claims = Claims {
        iss: client_id.clone(),
        sub: client_id,
        aud: token_url.clone(),
        iat: now,
        exp: now + 300,
    };

    let mut header = Header::new(Algorithm::ES256);
    header.typ = Some("JWT".into());
    let assertion = encode(&header, &claims, &key)?;

    let resp = reqwest::Client::new()
        .post(&token_url)
        .form(&[
            ("grant_type", "client_credentials"),
            (
                "client_assertion_type",
                "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
            ),
            ("client_assertion", &assertion),
        ])
        .send()
        .await?;
    println!("{} {}", resp.status(), resp.text().await?);
    Ok(())
}