Verifying in other languages

import base64
import hashlib
import hmac
import json
import time

TOLERANCE_SECONDS = 300


def verify_signed_request(secret: str, raw_body: bytes, header: str, kind: str, now: int | None = None) -> dict:
    """Checks storeware-signature for an endpoint that serves `kind`. Raises ValueError if it fails."""
    parts = dict(p.strip().split("=", 1) for p in header.split(",") if "=" in p)
    t, v2 = parts.get("t"), parts.get("v2")
    if not t or not v2:  # v1 alone is never enough: it signs no kind
        raise ValueError("malformed signature header")
    timestamp = int(t)
    now = int(time.time()) if now is None else now
    if abs(now - timestamp) > TOLERANCE_SECONDS:
        raise ValueError("timestamp outside the tolerance window")
    signed = f"{timestamp}.{kind}.".encode() + raw_body
    expected = hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected.encode(), v2.encode()):
        raise ValueError("signature mismatch")
    return json.loads(raw_body)


def _b64url(part: str) -> bytes:
    return base64.urlsafe_b64decode(part + "=" * (-len(part) % 4))


def verify_session(token: str, secret: str, client_id: str, now: int | None = None) -> dict:
    """Checks an admin session token. Returns its claims, or raises ValueError."""
    parts = token.split(".")
    if len(parts) != 3:
        raise ValueError("malformed token")
    header, body, signature = parts
    decoded = json.loads(_b64url(header))
    if not isinstance(decoded, dict) or decoded.get("alg") != "HS256":
        raise ValueError("unsupported signature algorithm")
    digest = hmac.new(secret.encode(), f"{header}.{body}".encode(), hashlib.sha256).digest()
    expected = base64.urlsafe_b64encode(digest).rstrip(b"=")
    if not hmac.compare_digest(expected, signature.encode()):
        raise ValueError("signature mismatch")
    claims = json.loads(_b64url(body))
    now = int(time.time()) if now is None else now
    if claims.get("aud") != client_id:
        raise ValueError("token is for another app")
    if not isinstance(claims.get("exp"), int) or claims["exp"] <= now:
        raise ValueError("token has expired")
    if isinstance(claims.get("iat"), int) and claims["iat"] > now + 60:
        raise ValueError("token was issued in the future")
    return claims

<?php

const STOREWARE_TOLERANCE_SECONDS = 300;

/** Checks storeware-signature for an endpoint that serves $kind. Throws if it fails. */
function storeware_verify_signed_request(string $secret, string $rawBody, string $header, string $kind, ?int $now = null): array
{
    $parts = [];
    foreach (explode(',', $header) as $part) {
        $pair = explode('=', trim($part), 2);
        if (count($pair) === 2) {
            $parts[$pair[0]] = $pair[1];
        }
    }
    // v1 alone is never enough: it signs no kind.
    if (($parts['t'] ?? '') === '' || ($parts['v2'] ?? '') === '') {
        throw new RuntimeException('malformed signature header');
    }
    $timestamp = (int) $parts['t'];
    $now ??= time();
    if (abs($now - $timestamp) > STOREWARE_TOLERANCE_SECONDS) {
        throw new RuntimeException('timestamp outside the tolerance window');
    }
    $expected = hash_hmac('sha256', "{$timestamp}.{$kind}.{$rawBody}", $secret);
    if (!hash_equals($expected, $parts['v2'])) {
        throw new RuntimeException('signature mismatch');
    }
    return json_decode($rawBody, true, 512, JSON_THROW_ON_ERROR);
}

function storeware_base64url_decode(string $part): string
{
    return (string) base64_decode(strtr($part, '-_', '+/'), true);
}

/** Checks an admin session token. Returns its claims, or throws. */
function storeware_verify_session(string $token, string $secret, string $clientId, ?int $now = null): array
{
    $parts = explode('.', $token);
    if (count($parts) !== 3) {
        throw new RuntimeException('malformed token');
    }
    [$header, $body, $signature] = $parts;
    $decoded = json_decode(storeware_base64url_decode($header), true);
    if (!is_array($decoded) || ($decoded['alg'] ?? null) !== 'HS256') {
        throw new RuntimeException('unsupported signature algorithm');
    }
    $digest = hash_hmac('sha256', "{$header}.{$body}", $secret, true);
    $expected = rtrim(strtr(base64_encode($digest), '+/', '-_'), '=');
    if (!hash_equals($expected, $signature)) {
        throw new RuntimeException('signature mismatch');
    }
    $claims = json_decode(storeware_base64url_decode($body), true, 512, JSON_THROW_ON_ERROR);
    $now ??= time();
    if (($claims['aud'] ?? null) !== $clientId) {
        throw new RuntimeException('token is for another app');
    }
    if (!is_int($claims['exp'] ?? null) || $claims['exp'] <= $now) {
        throw new RuntimeException('token has expired');
    }
    if (is_int($claims['iat'] ?? null) && $claims['iat'] > $now + 60) {
        throw new RuntimeException('token was issued in the future');
    }
    return $claims;
}

package storeware

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"strconv"
	"strings"
)

const toleranceSeconds = 300

// VerifySignedRequest checks storeware-signature for an endpoint that serves kind.
// rawBody is the body as read, before any JSON decoding; now is Unix seconds.
func VerifySignedRequest(secret string, rawBody []byte, header, kind string, now int64) error {
	parts := map[string]string{}
	for _, part := range strings.Split(header, ",") {
		if key, value, ok := strings.Cut(strings.TrimSpace(part), "="); ok {
			parts[key] = value
		}
	}
	// v1 alone is never enough: it signs no kind.
	t, v2 := parts["t"], parts["v2"]
	if t == "" || v2 == "" {
		return errors.New("malformed signature header")
	}
	timestamp, err := strconv.ParseInt(t, 10, 64)
	if err != nil {
		return errors.New("malformed timestamp")
	}
	if age := now - timestamp; age > toleranceSeconds || age < -toleranceSeconds {
		return errors.New("timestamp outside the tolerance window")
	}
	mac := hmac.New(sha256.New, []byte(secret))
	fmt.Fprintf(mac, "%d.%s.", timestamp, kind)
	mac.Write(rawBody)
	expected := hex.EncodeToString(mac.Sum(nil))
	if !hmac.Equal([]byte(expected), []byte(v2)) {
		return errors.New("signature mismatch")
	}
	return nil
}

type SessionClaims struct {
	Iss     string         `json:"iss"`
	Aud     string         `json:"aud"`
	Sub     string         `json:"sub"`
	Exp     int64          `json:"exp"`
	Iat     int64          `json:"iat"`
	Jti     string         `json:"jti"`
	Sid     string         `json:"sid"`
	StoreID string         `json:"store_id"`
	OrgID   string         `json:"org_id"`
	Context map[string]any `json:"context"`
}

// VerifySession checks an admin session token and returns its claims.
func VerifySession(token, secret, clientID string, now int64) (*SessionClaims, error) {
	parts := strings.Split(token, ".")
	if len(parts) != 3 {
		return nil, errors.New("malformed token")
	}
	var header struct {
		Alg string `json:"alg"`
	}
	headerJSON, err := base64.RawURLEncoding.DecodeString(parts[0])
	if err != nil || json.Unmarshal(headerJSON, &header) != nil || header.Alg != "HS256" {
		return nil, errors.New("unsupported signature algorithm")
	}
	mac := hmac.New(sha256.New, []byte(secret))
	mac.Write([]byte(parts[0] + "." + parts[1]))
	expected := base64.RawURLEncoding.EncodeToString(mac.Sum(nil))
	if !hmac.Equal([]byte(expected), []byte(parts[2])) {
		return nil, errors.New("signature mismatch")
	}
	var claims SessionClaims
	body, err := base64.RawURLEncoding.DecodeString(parts[1])
	if err != nil || json.Unmarshal(body, &claims) != nil {
		return nil, errors.New("malformed claims")
	}
	if claims.Aud != clientID {
		return nil, errors.New("token is for another app")
	}
	if claims.Exp <= now {
		return nil, errors.New("token has expired")
	}
	if claims.Iat > now+60 {
		return nil, errors.New("token was issued in the future")
	}
	return &claims, nil
}

require "base64"
require "json"
require "openssl"

TOLERANCE_SECONDS = 300

# Checks storeware-signature for an endpoint that serves kind. Raises if it fails.
def verify_signed_request(secret, raw_body, header, kind, now: Time.now.to_i)
  parts = header.split(",").map { |part| part.strip.split("=", 2) }.select { |pair| pair.size == 2 }.to_h
  t, v2 = parts["t"], parts["v2"]
  # v1 alone is never enough: it signs no kind.
  raise ArgumentError, "malformed signature header" if t.to_s.empty? || v2.to_s.empty?
  timestamp = Integer(t, 10)
  raise ArgumentError, "timestamp outside the tolerance window" if (now - timestamp).abs > TOLERANCE_SECONDS
  expected = OpenSSL::HMAC.hexdigest("SHA256", secret, "#{timestamp}.#{kind}.#{raw_body}")
  raise ArgumentError, "signature mismatch" unless OpenSSL.secure_compare(expected, v2)
  JSON.parse(raw_body)
end

# Checks an admin session token. Returns its claims, or raises.
def verify_session(token, secret, client_id, now: Time.now.to_i)
  parts = token.split(".", -1) # -1 keeps an empty last part
  raise ArgumentError, "malformed token" unless parts.size == 3
  header, body, signature = parts
  decoded = JSON.parse(Base64.urlsafe_decode64(header))
  raise ArgumentError, "unsupported signature algorithm" unless decoded.is_a?(Hash) && decoded["alg"] == "HS256"
  digest = OpenSSL::HMAC.digest("SHA256", secret, "#{header}.#{body}")
  expected = Base64.urlsafe_encode64(digest, padding: false)
  raise ArgumentError, "signature mismatch" unless OpenSSL.secure_compare(expected, signature)
  claims = JSON.parse(Base64.urlsafe_decode64(body))
  raise ArgumentError, "token is for another app" unless claims["aud"] == client_id
  raise ArgumentError, "token has expired" unless claims["exp"].is_a?(Integer) && claims["exp"] > now
  raise ArgumentError, "token was issued in the future" if claims["iat"].is_a?(Integer) && claims["iat"] > now + 60
  claims
end

import json
import sys

from storeware_verify import verify_session, verify_signed_request

vectors = json.load(open(sys.argv[1], encoding="utf-8"))
failed = 0
for v in vectors["signed_requests"]["vectors"] + vectors["session_tokens"]["vectors"]:
    try:
        if "token" in v:
            verify_session(v["token"], v["secret"], v["client_id"], v["now"])
        else:
            verify_signed_request(v["secret"], v["body"].encode(), v["headers"]["storeware-signature"], v["endpoint_kind"], v["now"])
        got = "accept"
    except ValueError:
        got = "refuse"
    failed += got != v["expect"]
    print("ok  " if got == v["expect"] else "FAIL", v["name"])
sys.exit(1 if failed else 0)

curl https://api.storeware.ai/api/oauth/token \
  --data-urlencode grant_type=urn:ietf:params:oauth:grant-type:token-exchange \
  --data-urlencode subject_token="$SESSION_TOKEN" \
  --data-urlencode subject_token_type=urn:storeware:params:oauth:token-type:session-token \
  --data-urlencode client_id="$CLIENT_ID" \
  --data-urlencode client_secret="$CLIENT_SECRET"