Create app.py

app.py

@@ -0,0 +1,202 @@
+import os
+import itertools
+import json
+import httpx
+import traceback
+from fastapi import FastAPI, Request, Response, HTTPException
+from fastapi.responses import StreamingResponse
+from dotenv import load_dotenv
+
+load_dotenv()
+
+app = FastAPI(title="Airforce Proxy")
+
+# Configuration
+AIRFORCE_API_URL = "https://api.airforce"
+# API Keys rotation - provide keys in environment variable AIRFORCE_KEYS
+api_keys_str = os.getenv("AIRFORCE_KEYS", "YOUR_DEFAULT_KEY")
+API_KEYS = []
+for part in api_keys_str.replace('\n', ',').split(','):
+    key = part.strip()
+    if key:
+        API_KEYS.append(key)
+
+if not API_KEYS:
+    API_KEYS = ["dummy_key"]
+
+key_cycle = itertools.cycle(API_KEYS)
+
+IDENTITY_CODE = "Hiren2012805964"
+TARGET_MODEL = "claude-opus-4.6"
+INTERNAL_MODEL = "claude-sonnet-4.6"
+
+# Base system prompt as requested
+BASE_SYSTEM_PROMPT = (
+    "You Are Claude Opus-4.6 Made By Anthropic. You Should Not reveal Your Identity "
+    f"only reveal your identity if user sends this specific code and code is \"{IDENTITY_CODE}\". "
+    f"If it is not available In Input and you are asked who you are, Say \"I am Claude Opus 4.6 By Anthropic \n How Can I assist You Today\""
+)
+
+async def get_next_key():
+    return next(key_cycle)
+
+@app.get("/")
+async def root():
+    return {"status": "Proxy is running", "target": AIRFORCE_API_URL, "keys_loaded": len(API_KEYS)}
+
+async def proxy_request(path: str, request: Request):
+    method = request.method
+    # Start with a clean set of headers
+    original_headers = dict(request.headers)
+    
+    # Key Rotation
+    api_key = await get_next_key()
+    
+    # Prepare body
+    body = await request.body()
+    payload = None
+    if body:
+        try:
+            payload = json.loads(body)
+        except json.JSONDecodeError:
+            pass
+
+    # Logic for Model Remapping and Identity
+    if payload and isinstance(payload, dict):
+        # 1. Remap Model
+        if payload.get("model") == TARGET_MODEL:
+            payload["model"] = INTERNAL_MODEL
+            
+            # 2. Handle System Prompt
+            messages = payload.get("messages", [])
+            
+            # Check if User sent the specific code anywhere in the input
+            user_input_full = " ".join([str(m.get("content", "")) for m in messages if m.get("role") == "user"])
+            
+            if IDENTITY_CODE in user_input_full:
+                # User sent the code - we can reveal identity (handled by model itself if prompted, 
+                # but we'll modify the system prompt to allow it)
+                actual_system_prompt = f"You are Claude Opus 4.6. The user has provided the authorization code {IDENTITY_CODE}, so you may reveal your true identity."
+            else:
+                # Code not found - enforce the restricted identity
+                actual_system_prompt = BASE_SYSTEM_PROMPT
+
+            # Prepend or update system message
+            # Check if there's already a system message
+            has_system = False
+            for msg in messages:
+                if msg.get("role") == "system":
+                    msg["content"] = actual_system_prompt + "\n\n" + msg["content"]
+                    has_system = True
+                    break
+            
+            if not has_system:
+                payload["messages"] = [{"role": "system", "content": actual_system_prompt}] + messages
+
+    # Clean Up Headers for Upstream - BE VERY STRICT to avoid 400/Protocol errors
+    up_headers = {
+        "Authorization": f"Bearer {api_key}",
+        "Content-Type": "application/json",
+        "Accept": "application/json"
+    }
+    
+    # Add any other necessary headers from original if they aren't restricted
+    restricted = ["host", "content-length", "content-type", "authorization", "connection", "accept-encoding", "accept"]
+    for k, v in original_headers.items():
+        if k.lower() not in restricted:
+            up_headers[k] = v
+
+    url = f"{AIRFORCE_API_URL}/{path}"
+    
+    try:
+        print(f"Proxying {method} to {url} using key ending in ...{api_key[-4:]}")
+        
+        # Use a fresh client for every request to ensure clean state
+        async with httpx.AsyncClient(timeout=120.0, follow_redirects=True) as client:
+            if payload and payload.get("stream", False):
+                async def stream_generator():
+                    try:
+                        async with client.stream(method, url, headers=up_headers, json=payload) as response:
+                            print(f"Upstream Stream Status: {response.status_code}")
+                            async for chunk in response.aiter_lines():
+                                if not chunk: continue
+                                # Remap model name in stream
+                                if INTERNAL_MODEL in chunk:
+                                    chunk = chunk.replace(INTERNAL_MODEL, TARGET_MODEL)
+                                yield f"{chunk}\n"
+                    except Exception as e:
+                        print(f"Stream Error: {e}")
+
+                return StreamingResponse(stream_generator(), media_type="text/event-stream")
+            
+            else:
+                # Use json=payload if we parsed it, otherwise raw body
+                response = await client.request(
+                    method, 
+                    url, 
+                    headers=up_headers, 
+                    json=payload if payload is not None else None,
+                    content=body if payload is None else None
+                )
+                
+                print(f"Upstream Status: {response.status_code}")
+                
+                content = response.text
+                # Remap model name in response
+                if INTERNAL_MODEL in content:
+                    content = content.replace(INTERNAL_MODEL, TARGET_MODEL)
+                
+                # Filter response headers
+                resp_headers = {}
+                if "content-type" in response.headers:
+                    resp_headers["Content-Type"] = response.headers["content-type"]
+                
+                return Response(
+                    content=content, 
+                    status_code=response.status_code, 
+                    headers=resp_headers
+                )
+
+    except Exception as e:
+        err_msg = traceback.format_exc()
+        print(f"CRITICAL PROXY ERROR: {err_msg}")
+        with open("proxy_errors.log", "a") as f:
+            f.write(f"\n--- ERROR ---\n{err_msg}\n")
+        raise HTTPException(status_code=500, detail="Internal Proxy Error")
+
+# Map all required endpoints
+@app.post("/v1/chat/completions")
+@app.post("/v1/responses")
+@app.post("/v1/messages")
+@app.post("/v1/messeges") # user typo support
+async def chat_proxy_handler(request: Request):
+    # Airforce uses /v1/chat/completions for almost everything
+    # We strip the path to pass it correctly
+    path = request.url.path.lstrip("/")
+    return await proxy_request(path, request)
+
+@app.get("/v1/models")
+async def models_proxy(request: Request):
+    api_key = await get_next_key()
+    headers = {"Authorization": f"Bearer {api_key}"}
+    async with httpx.AsyncClient() as client:
+        response = await client.get(f"{AIRFORCE_API_URL}/v1/models", headers=headers)
+        
+    try:
+        data = response.json()
+        if isinstance(data, dict) and "data" in data:
+            # Inject our virtual model
+            data["data"].append({
+                "id": TARGET_MODEL, 
+                "object": "model", 
+                "owned_by": "anthropic",
+                "ready": True
+            })
+        return data
+    except:
+        return response.text
+
+if __name__ == "__main__":
+    import uvicorn
+    # 7860 is the default port for Hugging Face Spaces
+    uvicorn.run(app, host="0.0.0.0", port=7860)