wenlucer 0.1.1

A developer-first API secret proxy — authenticate users, hide API keys, rate-limit, log and forward requests to any AI provider.

🔐 wenlucer

Generic API key proxy. Register any external API with its real key. Collaborators receive a parallel key (wl_live_...) and your proxy URL. They call your proxy exactly as they would the real API — but the real key is never exposed.

collaborator → (parallel key) → wenlucer → (real key) → stripe / openai / github / ...

pip install wenlucer

---

Why?

You want to give a contractor or team member access to your Stripe, OpenAI, or GitHub account — but you don't want to share the real API key. With wenlucer you share a proxy URL + a disposable parallel key. You control rate limits, expiry, and which APIs they can reach. The real key never leaves your server.

---

Storage

By default wenlucer stores everything in memory — if the process restarts, all tokens and API registrations are lost and collaborators lose access.

Enable persistence with one line:

proxy = SecretProxy(storage="wenlucer_state.json")

The file is created with chmod 600 automatically. It contains real API keys in plaintext — treat it exactly like a .env file:

echo "wenlucer_state.json" >> .gitignore   # never commit it

State is restored on restart: existing tokens (including expiry, rate limits, and allowed APIs) come back exactly as they were. Rate-limit counters reset on restart, which is acceptable — the windows are short-lived by design.

---

Quick start

Operator (you)

from wenlucer import SecretProxy

proxy = SecretProxy()

# Register any external API with its real credentials
proxy.add_api("stripe",    "https://api.stripe.com",     "sk_live_...", "bearer")
proxy.add_api("openai",    "https://api.openai.com",      "sk-...",      "bearer")
proxy.add_api("github",    "https://api.github.com",      "ghp_...",     "bearer")

# Create a parallel key for a collaborator
key = proxy.create_key(
    "alice",
    limits     = {"requests_per_day": 500, "requests_per_minute": 20},
    expires_in = "30d",
    allowed_apis = ["stripe"],          # restrict to specific APIs
)
print(key)   # wl_live_xxxxxxxxxxxxxxxxxxxx

proxy.run(port=8000)

Collaborator (them)

They receive only: http://your-host:8000 and wl_live_...

import httpx

# Works identically to calling Stripe directly — just different base_url + key
client = httpx.Client(
    base_url = "http://your-host:8000/stripe",
    headers  = {"Authorization": "Bearer wl_live_..."},
)
r = client.post("/v1/charges", json={"amount": 2000, "currency": "usd", ...})

# Or with curl
curl http://your-host:8000/stripe/v1/charges \
  -H "Authorization: Bearer wl_live_..." \
  -d amount=2000 -d currency=usd ...

wenlucer swaps the parallel key for the real Stripe key and forwards the request.

---

add_api — supporting any API

# Bearer token (most APIs)
proxy.add_api("openai",    "https://api.openai.com",      "sk-...",     "bearer")
proxy.add_api("stripe",    "https://api.stripe.com",      "sk_live...", "bearer")
proxy.add_api("github",    "https://api.github.com",      "ghp_...",    "bearer")
proxy.add_api("groq",      "https://api.groq.com",        "gsk_...",    "bearer")
proxy.add_api("mistral",   "https://api.mistral.ai",      "...",        "bearer")
proxy.add_api("twilio",    "https://api.twilio.com",      "ACxxx:tok",  "bearer")

# Custom header
proxy.add_api("anthropic", "https://api.anthropic.com",   "sk-ant-...",
              auth_style   = "header:x-api-key",
              extra_headers = {"anthropic-version": "2023-06-01"})

proxy.add_api("myapi",     "https://internal.corp.com",   "secret",
              auth_style   = "header:X-Internal-Token")

# Query-param key (e.g. Google Maps, some weather APIs)
proxy.add_api("maps",      "https://maps.googleapis.com", "AIza...",
              auth_style   = "query:key")

# Any custom REST API
proxy.add_api("payments",  "https://pay.acme.com/api",    "tok_live_...",
              auth_style   = "header:X-Payment-Key",
              extra_headers = {"X-Api-Version": "2024-01"})

The auth_style options:

Value	Result
"bearer"	Authorization: Bearer <real_key>
"header:Foo-Bar"	Foo-Bar: <real_key>
"query:api_key"	?api_key=<real_key> appended to every URL

---

Parallel key options

key = proxy.create_key(
    "contractor-xyz",
    limits = {
        "requests_per_minute": 10,
        "requests_per_hour":   200,
        "requests_per_day":    1000,
        "requests_per_month":  20_000,
    },
    expires_in   = "7d",               # s / m / h / d  — auto-expires
    allowed_apis = ["stripe", "openai"],  # restrict which APIs they can reach
    metadata     = {"team": "agency-abc", "contract": "project-42"},
)

Revoke or rotate at any time:

proxy.revoke_key(key)        # permanently invalidated
new_key = proxy.rotate_key(key)   # old key revoked, new one returned

---

URL mapping

For an API registered as "stripe", the proxy mounts everything under /stripe:

Client calls	Wenlucer forwards to
POST /stripe/v1/charges	https://api.stripe.com/v1/charges
GET /stripe/v1/customers/cus_xxx	https://api.stripe.com/v1/customers/cus_xxx
POST /openai/v1/chat/completions	https://api.openai.com/v1/chat/completions
GET /github/repos/owner/repo	https://api.github.com/repos/owner/repo

The prefix is stripped and the rest is forwarded verbatim — including path, query string, and body.

---

Streaming

Works transparently. Pass "stream": true in the body and wenlucer uses a StreamingResponse automatically.

---

Stats & logs

proxy.stats()
# {
#   "total_requests": 84,
#   "errors": 0,
#   "avg_latency_ms": 241.3,
#   "by_provider": {"stripe": 60, "openai": 24},
#   "by_user": {"alice": 50, "bob": 34}
# }

proxy.logs(user="alice", limit=20)

---

Management endpoints (all require Bearer auth)

Endpoint	Description
GET /_wenlucer/health	Health + registered APIs (no auth required)
GET /_wenlucer/stats	Request stats scoped to the current key
GET /_wenlucer/logs	Recent request log scoped to the current key
GET /_wenlucer/me	Current key info + rate limit status
POST /_wenlucer/rotate	Rotate your own key

/_wenlucer/stats returns only the calling key's own usage — collaborators cannot see each other's traffic.

---

Production deployment

Always run wenlucer behind HTTPS. The proxy handles real API keys — plain HTTP exposes them to network eavesdroppers.

nginx example:

server {
    listen 443 ssl;
    server_name proxy.example.com;

    ssl_certificate     /etc/letsencrypt/live/proxy.example.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/proxy.example.com/privkey.pem;

    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host              $host;
        proxy_set_header X-Real-IP         $remote_addr;
        proxy_set_header X-Forwarded-Proto https;

        # Required for streaming responses (OpenAI, Anthropic…)
        proxy_buffering    off;
        proxy_cache        off;
        chunked_transfer_encoding on;
    }
}

Caddy (automatic TLS via Let's Encrypt):

proxy.example.com {
    reverse_proxy localhost:8000 {
        flush_interval -1   # required for streaming
    }
}

gunicorn / uvicorn programmatic (ASGI):

# server.py
from wenlucer import SecretProxy

proxy = SecretProxy(storage="wenlucer_state.json", admin_key="...")
proxy.add_api("stripe", "https://api.stripe.com", "sk_live_...", "bearer")
app = proxy.build()   # returns a FastAPI instance

# gunicorn server:app -w 4 -k uvicorn.workers.UvicornWorker
# uvicorn server:app --host 0.0.0.0 --port 8000 --workers 4

Checklist before going live:

• TLS termination in place (nginx / Caddy / cloud load balancer)
• storage file excluded from version control (.gitignore)
• storage file on an encrypted volume or backed by a secrets manager
• admin_key set to a strong random secret
• Collaborator keys have expires_in set
• Rate limits configured per key

---

Admin endpoints

Pass admin_key to enable HTTP management endpoints for the operator:

proxy = SecretProxy(
    admin_key = "your-secret-admin-key",
    storage   = "wenlucer_state.json",
)

Endpoint	Description
GET /_wenlucer/admin/keys	List all parallel keys (tokens truncated)
POST /_wenlucer/admin/keys	Create a new parallel key
POST /_wenlucer/admin/revoke	Revoke a parallel key
GET /_wenlucer/admin/stats	Global request stats across all users

All admin endpoints require Authorization: Bearer <admin_key>.

# List all keys
curl http://localhost:8000/_wenlucer/admin/keys \
  -H "Authorization: Bearer your-secret-admin-key"

# Create a key
curl http://localhost:8000/_wenlucer/admin/keys \
  -H "Authorization: Bearer your-secret-admin-key" \
  -H "Content-Type: application/json" \
  -d '{"user": "alice", "limits": {"requests_per_day": 500}, "expires_in": "30d", "allowed_apis": ["stripe"]}'

# Revoke a key
curl -X POST http://localhost:8000/_wenlucer/admin/revoke \
  -H "Authorization: Bearer your-secret-admin-key" \
  -H "Content-Type: application/json" \
  -d '{"key": "wl_live_..."}'

If admin_key is not set, all /_wenlucer/admin/* routes return 404.

---

CLI

# Start with persistence and admin endpoints enabled
wenlucer start \
  --api stripe=https://api.stripe.com=sk_live_xxx=bearer \
  --api openai=https://api.openai.com=sk-xxx=bearer \
  --storage wenlucer_state.json \
  --admin-key your-secret-admin-key \
  --port 8000

# Create a key via storage file (use when the server is stopped)
wenlucer key alice \
  --rpd 500 --rpm 20 --expires-in 30d \
  --storage wenlucer_state.json

# Create a key against a live server (use when the server is running)
wenlucer key alice \
  --rpd 500 --rpm 20 --expires-in 30d \
  --server http://localhost:8000 \
  --admin-key your-secret-admin-key

---

Architecture

collaborator
    │  Authorization: Bearer wl_live_...
    ▼
wenlucer (FastAPI + uvicorn)
    ├── validate parallel key
    ├── check rate limits
    ├── decrypt real API key (Fernet, in-memory only)
    ├── build upstream URL  (strip /name prefix)
    ├── inject real auth header (or query param)
    ├── forward via httpx  (streaming-aware)
    ├── log request
    └── persist state (optional, if storage= is set)
            │  Authorization: Bearer sk_live_...
            ▼
      upstream API (Stripe / OpenAI / GitHub / ...)

operator
    │  Authorization: Bearer <admin_key>
    ▼
/_wenlucer/admin/*  (list keys, create, revoke, global stats)

---

Project structure

src/wenlucer/
├── __init__.py       public API
├── core.py           SecretProxy + FastAPI app + _Route
├── secrets.py        Fernet encrypted in-memory store
├── tokens.py         wl_live_... key generation & validation
├── limits.py         rate limiting (minute / hour / day / month)
├── logger.py         structured request log
├── persistence.py    optional JSON state file
├── client.py         WenluClient + AsyncWenluClient
└── cli.py            wenlucer CLI

---

License

Creative Commons Attribution-NoDerivatives 4.0 International (CC BY-ND 4.0). You are free to use and share this software, even for commercial purposes, as long as you provide attribution and do not distribute modified versions.

---

WenluClient — collaborator SDK

The collaborator receives only the proxy URL and their parallel key. WenluClient injects the key automatically on every request — no other configuration needed.

Basic usage (sync)

from wenlucer import WenluClient

client = WenluClient(
    base_url     = "http://your-server:8000",  # wenlucer proxy URL
    parallel_key = "wl_live_...",              # key issued by the operator
)

# GET
r = client.get("github", "/repos/owner/repo")
print(r.json())

# POST with JSON (identical to calling the real API)
r = client.post("openai", "/v1/chat/completions", json={
    "model": "gpt-4o",
    "messages": [{"role": "user", "content": "Hello!"}],
})
print(r.json()["choices"][0]["message"]["content"])

# POST with form data (Stripe style)
r = client.post("stripe", "/v1/charges", data={
    "amount": "2000",
    "currency": "usd",
    "source": "tok_visa",
})

# PUT / PATCH / DELETE also available
client.put("myapi", "/v1/resource/123", json={"status": "active"})
client.delete("myapi", "/v1/resource/123")

Streaming (OpenAI, Anthropic, etc.)

for chunk in client.stream("POST", "openai", "/v1/chat/completions", json={
    "model": "gpt-4o",
    "messages": [{"role": "user", "content": "Write me a poem"}],
    "stream": True,
}):
    print(chunk, end="", flush=True)

Introspection

# Available APIs on the server (no auth required)
print(client.health())
# {"status": "ok", "apis": {"openai": "https://api.openai.com", ...}}

# Current key info + rate-limit usage
info = client.me()
print(info["user"])          # your identifier
print(info["rate_limits"])   # current usage vs. limits
print(info["expires_at"])    # when the key expires

# Your last 20 requests
logs = client.my_logs(limit=20)
for entry in logs:
    print(entry["method"], entry["endpoint"], entry["status_code"])

# Rotate your own key (client updates itself automatically)
new_key = client.rotate_key()
print("New key:", new_key)

Context manager

with WenluClient("http://your-server:8000", "wl_live_...") as client:
    r = client.get("github", "/user")
    print(r.json())

Async usage

import asyncio
from wenlucer import AsyncWenluClient

async def main():
    client = AsyncWenluClient(
        base_url     = "http://your-server:8000",
        parallel_key = "wl_live_...",
    )

    r = await client.post("openai", "/v1/chat/completions", json={
        "model": "gpt-4o",
        "messages": [{"role": "user", "content": "Hello!"}],
    })
    print(r.json())

    async for chunk in client.stream("POST", "openai", "/v1/chat/completions",
                                     json={..., "stream": True}):
        print(chunk, end="", flush=True)

    info = await client.me()
    print(info)

asyncio.run(main())

Error handling

from wenlucer import WenluClient, WenluClientError

client = WenluClient(...)

try:
    r = client.post("openai", "/v1/chat/completions", json={...})
except WenluClientError as e:
    # 401 — invalid or expired key
    # 403 — API not allowed for this key
    # 429 — rate limit reached
    print("Proxy error:", e)

Client parameters

Parameter	Type	Default	Description
base_url	str	—	Root URL of the wenlucer server
parallel_key	str	—	wl_live_... key issued by the operator
timeout	float	60.0	Per-request timeout in seconds
verify_ssl	bool	True	Verify SSL certificate