koios_api_header_test.sh

koios_api_header_test.sh

#!/bin/bash

# Use this script to test the two variations of Koios API headers
# The script will test the headers by continuously and concurrently querying the Koios API
# Allow the test to run 5000 times in order to reach the Koios rate limit of the public tier

# Usage:
# Test the broken headers: ./koios_api_header_test.sh broken
# Test the working headers: ./koios_api_header_test.sh working

# Broken headers are the ones as defined in the env file:
# KOIOS_API_HEADERS=$($CURL -s "${KOIOS_API_HEADERS[@]}" "${URL}"

# Working headers are the ones as defined in the proposed changes to the new heathcheck:
# KOIOS_API_HEADERS=$(${CURL} -s ${KOIOS_API_TOKEN:+-H "Authorization: Bearer ${KOIOS_API_TOKEN}"} ${URL})

KOIOS_API_TOKEN="<your_token_here>"
MAX_CONCURRENT=10 # Define the maximum number of concurrent processes

URL="https://api.koios.rest/api/v1/tip" # Mainnet
CURL=$(which curl)
JQ=$(which jq)

if [[ -z "${KOIOS_API_HEADERS[*]}" ]]; then
    if [[ -n "${KOIOS_API_TOKEN}" ]]; then
        KOIOS_API_HEADERS=(-H "'Authorization: Bearer ${KOIOS_API_TOKEN}'")
    else
        KOIOS_API_HEADERS=()
    fi
fi

# Function to wait for background processes to finish
wait_for_jobs() {
    while [ $(jobs -r | wc -l) -ge $MAX_CONCURRENT ]; do
        sleep 0.1
    done
}

# Function to print the test results
print_test_results() {
    local test_number=$1
    local koios_response=$2
    local command=$4
    local current_height=$5

    if [[ -z "$current_height" ]]; then
        echo -e "\n----------------------------------------"
        echo -e "Error. Test ${test_number} failed"
    else
        echo -e "\n----------------------------------------"
        echo -e "\nTest ${test_number} passed"
        echo -e "\nHeight reported by Koios API:\n${current_height}"
    fi
    echo -e "\nKoios output:\n\"${koios_response}\""
    echo -e "\nCommand used:\n${command}"
}

# Function to run the tests based on the argument
run_header_test() {
    local test_type=$1
    local koios_response_command

    if [[ "$test_type" == "broken" ]]; then
        koios_response_command="$CURL -s \"${KOIOS_API_HEADERS[*]}\" \"${URL}\""
    elif [[ "$test_type" == "working" ]]; then
        koios_response_command="$CURL -s ${KOIOS_API_TOKEN:+-H \"Authorization: Bearer ${KOIOS_API_TOKEN}\"} \"${URL}\""
    else
        echo "Invalid argument. Use 'broken' or 'working'."
        exit 1
    fi

    # Test runs 5k+ times to get above the koios public daily tier rate limit
    for i in {1..5010}; do
        wait_for_jobs # Wait if the number of concurrent jobs reaches the limit
        {
            # Query the Koios API
            KOIOS_RESPONSE=$(eval $koios_response_command)
            # Check if the response is valid json
            if ! echo ${KOIOS_RESPONSE} | $JQ . &>/dev/null; then
                print_test_results $i "${KOIOS_RESPONSE}" "${KOIOS_API_HEADERS[*]}" "${koios_response_command}"
            else
                CURRENT_HEIGHT=$(echo "${KOIOS_RESPONSE}" | $JQ '.[0].block_no')
                print_test_results $i "${KOIOS_RESPONSE}" "${KOIOS_API_HEADERS[*]}" "${koios_response_command}" "$CURRENT_HEIGHT"
            fi
        } & # Run the block in the background

        # Wait for all background jobs to finish
        wait
    done
}

# Call the function to run the tests with the provided argument
run_header_test "$1"

Here is the python script I mentioned that I used to examine the headers:

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from fastapi.middleware.cors import CORSMiddleware
import uvicorn
import time
from collections import defaultdict

app = FastAPI()

# Simple in-memory store: {ip: [timestamps]}
rate_limit_store = defaultdict(list)
RATE_LIMIT = 20  # requests
TIME_WINDOW = 60  # seconds

# Your single valid API token
API_TOKEN = "AbC1234567890xYzABCd"  # 20-character token

@app.middleware("http")
async def rate_limiter(request: Request, call_next):
    headers = dict(request.headers)
    print("Received Headers:")
    for key, value in headers.items():
        print(f"{key}: {value}")

    auth_header = headers.get("authorization", "")
    if auth_header.lower().startswith("bearer "):
        token = auth_header[7:]
        if token == API_TOKEN:
            return await call_next(request)

    # If not authenticated, apply rate limiting per IP
    client_ip = request.client.host
    now = time.time()
    request_times = rate_limit_store[client_ip]

    # Keep only recent requests within the time window
    request_times = [t for t in request_times if now - t < TIME_WINDOW]
    rate_limit_store[client_ip] = request_times

    if len(request_times) >= RATE_LIMIT:
        raise HTTPException(status_code=429, detail="Too Many Requests")

    # Log the request
    rate_limit_store[client_ip].append(now)
    return await call_next(request)

@app.get("/")
async def read_root():
    return JSONResponse(content={"message": "Hello, world!", "status": "success"})

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)