AWS SigV4 Authentication

import { defaultProvider } from "@aws-sdk/credential-provider-node";
import { HttpRequest } from "@aws-sdk/protocol-http";
import { Sha256 } from '@aws-crypto/sha256-js';
import { SignatureV4 } from "@aws-sdk/signature-v4";
import { stringify } from "qs";

// Create a presigned STS GetCallerIdentity URL
async function createPresignedURL(): Promise<string> {
  const region = process.env.AWS_REGION || "us-east-1";

  const signer = new SignatureV4({
    credentials: defaultProvider(),
    region: region,
    service: "sts",
    sha256: Sha256,
  });

  const host = `sts.${region}.amazonaws.com`;
  const req = new HttpRequest({
    headers: { 'Content-Type': 'application/json', host },
    hostname: host,
    method: "GET",
    path: '/',
    query: {
      Action: 'GetCallerIdentity',
      Version: '2011-06-15',
    },
  });

  const signed = await signer.presign(req, { expiresIn: 60 * 10 });
  return `https://${signed.hostname}${signed.path}?${stringify(signed.query)}`;
}

// Make authenticated request to Formal API
async function callFormalAPI(endpoint: string, body: any) {
  const presignedURL = await createPresignedURL();
  const url = new URL(endpoint, 'https://api.joinformal.com');

  const response = await fetch(url.toString(), {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'Authorization': `AWS4-Presigned-URL ${presignedURL}`
    },
    body: JSON.stringify(body),
  });

  return response.json();
}

// Example: List resources
const resources = await callFormalAPI(
  '/core.v1.ResourceService/ListResources',
  { limit: 100 }
);

import boto3
import requests
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest

# Create a presigned STS GetCallerIdentity URL
def create_presigned_url():
    session = boto3.Session()
    credentials = session.get_credentials()
    region = session.region_name or 'us-east-1'

    url = f'https://sts.{region}.amazonaws.com/?Action=GetCallerIdentity&Version=2011-06-15'
    request = AWSRequest(method='GET', url=url)
    SigV4Auth(credentials, 'sts', region).add_auth(request)

    return request.url

# Make authenticated request to Formal API
def call_formal_api(endpoint, body):
    presigned_url = create_presigned_url()
    url = f'https://api.joinformal.com{endpoint}'

    response = requests.post(
        url,
        json=body,
        headers={
            'Content-Type': 'application/json',
            'Authorization': f'AWS4-Presigned-URL {presigned_url}'
        }
    )

    return response.json()

# Example: List resources
resources = call_formal_api(
    '/core.v1.ResourceService/ListResources',
    {'limit': 100}
)

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "net/http"
    "time"
    "bytes"
    "io"

    "github.com/aws/aws-sdk-go-v2/config"
    "github.com/aws/aws-sdk-go-v2/service/sts"
)

// Create a presigned STS GetCallerIdentity URL
func createPresignedURL(ctx context.Context) (string, error) {
    cfg, err := config.LoadDefaultConfig(ctx)
    if err != nil {
        return "", err
    }

    stsClient := sts.NewFromConfig(cfg)
    presignClient := sts.NewPresignClient(stsClient)

    presignedReq, err := presignClient.PresignGetCallerIdentity(ctx, &sts.GetCallerIdentityInput{},
        func(opts *sts.PresignOptions) {
            opts.Expires = 10 * time.Minute
        })
    if err != nil {
        return "", err
    }

    return presignedReq.URL, nil
}

// Make authenticated request to Formal API
func callFormalAPI(ctx context.Context, endpoint string, body interface{}) ([]byte, error) {
    presignedURL, err := createPresignedURL(ctx)
    if err != nil {
        return nil, err
    }

    apiURL := "https://api.joinformal.com" + endpoint

    bodyBytes, _ := json.Marshal(body)
    req, _ := http.NewRequestWithContext(ctx, "POST", apiURL, bytes.NewReader(bodyBytes))
    req.Header.Set("Content-Type", "application/json")
    req.Header.Set("Authorization", fmt.Sprintf("AWS4-Presigned-URL %s", presignedURL))

    resp, err := http.DefaultClient.Do(req)
    if err != nil {
        return nil, err
    }
    defer resp.Body.Close()

    return io.ReadAll(resp.Body)
}

// Example: List resources
func main() {
    ctx := context.Background()
    body := map[string]interface{}{"limit": 100}

    result, err := callFormalAPI(ctx, "/core.v1.ResourceService/ListResources", body)
    if err != nil {
        panic(err)
    }

    fmt.Println(string(result))
}

# Step 1: Generate presigned URL using AWS CLI
PRESIGNED_URL=$(aws sts get-caller-identity --output text --query 'Account' 2>/dev/null && \
  aws sts get-caller-identity --generate-cli-skeleton | \
  aws sts get-caller-identity --cli-input-json file:///dev/stdin --endpoint-url https://sts.us-east-1.amazonaws.com)

# For a simpler approach, use a script to generate the presigned URL
# Then use it with curl:

curl -X POST https://api.joinformal.com/core.v1.ResourceService/ListResources \
  -H "Content-Type: application/json" \
  -H "Authorization: AWS4-Presigned-URL <YOUR_PRESIGNED_URL>" \
  -d '{"limit": 100}'

# Example: Allow specific machine identities to list resources
package formal.app

default allow := false

allow if {
    input.request.path == "/core.v1.ResourceService/ListResources"
    input.user.name in ["DataPipeline", "ETLService"]
}