from pydantic import BaseModel from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded from fastapi.security.api_key import APIKeyHeader from fastapi.middleware.cors import CORSMiddleware from slowapi import Limiter, _rate_limit_exceeded_handler import base64, os, requests, paramiko, io, json, asyncio, subprocess from fastapi import FastAPI, Depends, HTTPException, Security, status, Header, BackgroundTasks, Request ############################################################################################################################ # Setup App and Standard Default Route app = FastAPI() @app.get("/") def greet_json(): return {"status": "online", "message": "All in One API is Active"} ############################################################################################################################ ################################################################################################################################### ############################################## Jarvis Initial Setup (Zcron) ####################################################### ################################################################################################################################### # --- REMOTE COMMAND FUNCTION --- def run_remote_cmd(command_string: str): url = "https://rjvishwa-testing.hf.space/execute" # Environment variable get the JLAB_KEY_ZCRON jlab_key_zcron = os.getenv("JLAB_KEY_ZCRON") if not jlab_key_zcron: raise ValueError("JLAB_KEY environment variable missing!") headers = {"X-API-Key": jlab_key_zcron} # Base64 Encoding cmd_b64 = base64.b64encode(command_string.encode('utf-8')).decode('utf-8') response = requests.post(url, headers=headers, json={"cmd_base64": cmd_b64}, timeout=60) return response.json() # --- TRIGGER ROUTE --- HARD_TO_GUESS_ROUTE_ZCRON = "/api/v1/cron-job/trigger-execution-x9f8a7b6c5d4e321-secure" # Setup the Jarvis via Cron Job @app.get(HARD_TO_GUESS_ROUTE_ZCRON) def run_cron_setup(): token = os.getenv("TOKEN_ZCRON") if not token: raise HTTPException(status_code=500, detail="TOKEN Environment Variable missing!") try: # Step 1: JarvisLabs Install Command install_res = run_remote_cmd("pip install jarvislabs==0.2.15") install_stdout = install_res.get("stdout", "No output") ################################################################################################################################ #install_res = subprocess.run(["pip", "install", "jarvislabs==0.2.15"], capture_output=True, text=True, check=True) #install_stdout = install_res.stdout or "No output" ################################################################################################################################ # Step 2: Setup Command with Token setup_cmd = f"jl setup --token '{token}' --yes" setup_res = run_remote_cmd(setup_cmd) setup_stdout = setup_res.get("stdout", "No output") ################################################################################################################################# #setup_res = subprocess.run(setup_cmd, shell=True, capture_output=True, text=True, check=True) #setup_stdout = setup_res.stdout or "No output" ################################################################################################################################# return {"status": "success", "message": "Cron execution completed!", "details": {"install_logs": install_stdout, "setup_logs": setup_stdout}} # Handle the Exception except Exception as e: raise HTTPException(status_code=500, detail=f"Error occurred: {str(e)}!") ################################################################################################################################### ########################################## Orion/Orix Public IP Setup (Orion/Orix) ################################################ ################################################################################################################################### # We use triple quotes to safely include single and double quotes ORION_COMMAND = """sudo useradd -m -s /bin/bash -G sudo orion && echo "orion:orion" | sudo chpasswd && sudo sed -i 's/^#\?PasswordAuthentication .*/PasswordAuthentication yes/; s/^#\?KbdInteractiveAuthentication .*/KbdInteractiveAuthentication yes/; s/^#\?ChallengeResponseAuthentication .*/ChallengeResponseAuthentication yes/; s/^#\?UsePAM .*/UsePAM yes/' /etc/ssh/sshd_config && sudo rm -f /etc/motd /run/motd.dynamic && sudo chmod -x /etc/update-motd.d/* && sudo systemctl restart ssh""" # type: ignore ORIX_COMMAND = """sudo rm -f /etc/motd /run/motd.dynamic && sudo chmod -x /etc/update-motd.d/*""" # type: ignore ORION_PASS_KEY = os.getenv("ORION_PASS_KEY") ORIX_PASS_KEY = os.getenv("ORION_PASS_KEY") # Orion IPV4 Setup: Use /deploy?ip=YOUR_IP e.g. "https://your-space-url.hf.space/deploy?ip=123.456.78.90" to run the script @app.get("/deploy") def deploy_to_server(ip: str, x_api_key: str = Header(...)): # Verify the Header Pass-Key if x_api_key != ORION_PASS_KEY: raise HTTPException(status_code=403, detail="Invalid Pass-Key.") # 1. Get Credentials from Hugging Face Secrets ssh_user = os.getenv("SSH_USER") ssh_key_str = os.getenv("SSH_KEY") if not ssh_key_str: raise HTTPException(status_code=500, detail="SSH_KEY secret is missing in environment variables.") client = paramiko.SSHClient() try: # 2. Try to Setup SSH Key # Convert string key to a file-like object for Paramiko key_file = io.StringIO(ssh_key_str) # Assuming the PEM is RSA. If it is a different format, change RSAKey to the appropriate type private_key = paramiko.RSAKey.from_private_key(key_file) # 3. Connect to the Server client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) print(f"Connecting to {ip} as {ssh_user}...") client.connect(hostname=ip, username=ssh_user, pkey=private_key, timeout=10) # 4. Execute the Command print(f"Running command on {ip}...") stdin, stdout, stderr = client.exec_command(ORION_COMMAND) # Wait for command to complete and read output exit_status = stdout.channel.recv_exit_status() output_msg = stdout.read().decode().strip() error_msg = stderr.read().decode().strip() # Close the remote ssh connection client.close() # 5. Return Json Result if exit_status == 0: return {"status": "success", "ip": ip, "message": "Script Executed successfully.", "output": output_msg} else: return {"status": "error", "ip": ip, "exit_code": exit_status, "error": error_msg} # Handle the Exception except Exception as e: return {"status": "connection_failed", "ip": ip, "details": str(e)} # Orix Deploy: Use /orix_deploy?ip=YOUR_IP&port=YOUR_PORT e.g. "https://your-space-url.hf.space/orix_deploy?ip=123.456.78.90&port=2222" @app.get("/orix_deploy") def deploy_orix_to_server(ip: str, port: int, user: str, x_api_key: str = Header(...)): # Verify the Header Pass-Key if x_api_key != ORIX_PASS_KEY: raise HTTPException(status_code=403, detail="Invalid Pass-Key.") # 1. Get Credentials from Hugging Face Secrets ssh_user = user ssh_key_str = os.getenv("SSH_KEY") if not ssh_key_str: raise HTTPException(status_code=500, detail="SSH_KEY secret is missing in environment variables.") client = paramiko.SSHClient() try: # 2. Try to Setup SSH Key key_file = io.StringIO(ssh_key_str) private_key = paramiko.RSAKey.from_private_key(key_file) # 3. Connect to the Server 'Using IP and PORT' client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) print(f"Connecting to {ip} on port {port} as {ssh_user}...") client.connect(hostname=ip, port=port, username=ssh_user, pkey=private_key, timeout=10) # 4. Execute the Command print(f"Running ORIX command on Instance with {ip}:{port}...") stdin, stdout, stderr = client.exec_command(ORIX_COMMAND) # Wait for command to complete and read output exit_status = stdout.channel.recv_exit_status() output_msg = stdout.read().decode().strip() error_msg = stderr.read().decode().strip() # Close the remote ssh connection client.close() # 5. Return Json Result if exit_status == 0: return {"status": "success", "ip": ip, "port": port, "message": "Orix Script Executed successfully.", "output": output_msg} else: return {"status": "error", "ip": ip, "port": port, "exit_code": exit_status, "error": error_msg} # Handle the Exception except Exception as e: return {"status": "connection_failed", "ip": ip, "port": port, "details": str(e)} ################################################################################################################################### ################################################# API/SDK For Jarvis (ZSRS) ####################################################### ################################################################################################################################### JLAB_KEY_ZSRS = os.getenv("JLAB_KEY_ZSRS") api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False) # Request Body Format class CommandPayload(BaseModel): cmd_base64: str # Verify the Pass Key async def verify_api_key(api_key_header: str = Security(api_key_header)): if api_key_header == JLAB_KEY_ZSRS: return api_key_header # Handle Exception raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Invalid API Key.") # Execute the command and provide the output @app.post("/execute") async def execute_any_command(payload: CommandPayload, api_key: str = Depends(verify_api_key)): command = base64.b64decode(payload.cmd_base64).decode('utf-8') process = await asyncio.create_subprocess_shell(command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE) stdout, stderr = await process.communicate() output_text = stdout.decode("utf-8").strip() try: return json.loads(output_text) except json.JSONDecodeError: return {"stdout": output_text, "stderr": stderr.decode("utf-8").strip()} ################################################################################################################################### ############################################### Delete Crunch Trash NVMe (Nova) ################################################### ################################################################################################################################### NOVA_PASSWD = os.getenv("NOVA_PASSWD") NOVA_CLIENT_ID = os.getenv("NOVA_CLIENT_ID") NOVA_BASE_URL = os.getenv("NOVA_DATACRUNCH_URL") NOVA_CLIENT_SECRET = os.getenv("NOVA_CLIENT_SECRET") # Verify the cron request before action def verify_cron_auth(x_cron_pass: str = Header(...)): if x_cron_pass != NOVA_PASSWD: raise HTTPException(status_code=401, detail="Unauthorized: Invalid cron Key/Password!") # Terminate Permanently the Trach NVMe Storage in the Crunch/Nova def background_cleanup_task(base_url: str, client_id: str, client_secret: str): try: # Start the Process... auth_url = f"{base_url}/v1/oauth2/token" auth_payload = {"grant_type": "client_credentials", "client_id": client_id, "client_secret": client_secret} auth_response = requests.post(auth_url, json=auth_payload) if auth_response.status_code != 200: print(f"Background Task Failed - Auth Error: {auth_response.text}!") return # Safely Exit Function access_token = auth_response.json().get("access_token") common_headers = {"Authorization": f"Bearer {access_token}", "Content-Type": "application/json"} trash_url = f"{base_url}/v1/volumes/trash" trash_response = requests.get(trash_url, headers=common_headers) if trash_response.status_code != 200: print(f"Background Task Failed - Trash Fetch Error: {trash_response.text}!") return # Safely Exit Function volumes_data = trash_response.json() if not volumes_data: return # Safely Exit Function for volume in volumes_data: vol_id = volume.get("id") if vol_id: delete_url = f"{base_url}/v1/volumes/{vol_id}" delete_payload = {"is_permanent": True} delete_response = requests.delete(delete_url, headers=common_headers, json=delete_payload) if delete_response.status_code in [200, 202, 204]: print(f"Successfully deleted: {vol_id}!") else: print(f"Error deleting {vol_id}: {delete_response.text}!") # If All NVMe Termination Successful print("Background cleanup finished successfully!") # Handle the Exception except Exception as e: print(f"Background Task Exception: {str(e)}.") @app.post("/cleanup-trash", dependencies=[Depends(verify_cron_auth)]) def cleanup_trash(background_tasks: BackgroundTasks): if not all([NOVA_BASE_URL, NOVA_CLIENT_ID, NOVA_CLIENT_SECRET]): raise HTTPException(status_code=500, detail="Missing API configuration in environment") # Do this process in background background_tasks.add_task(background_cleanup_task, NOVA_BASE_URL, NOVA_CLIENT_ID, NOVA_CLIENT_SECRET) # type: ignore # Return Success Json Response return {"status": "success", "message": "Trash cleanup process started in the background."} ################################################################################################################################### ############################################ Show and Cache the Pricing Data (Proxy) ############################################## # Initialize the rate limiter, We'll use the client's IP address as the key limiter = Limiter(key_func=get_remote_address) # Add the rate limiter to the app state app.state.limiter = limiter # Add a custom exception handler for rate limit exceeded errors app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) # type: ignore # --- CORS Middleware Setup --- # This allows your API to be called from web pages hosted on different domains # The settings below are very permissive, allowing any origin, method, and header app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"],) # --- API Endpoint Definition --- # This is the new route that will fetch and serve the pricing data @app.get("/pricing") # Apply the rate limit: 30 requests per minute per IP @limiter.limit("30/minute") def get_pricing_data(request: Request): # Load the password from the environment variable 'TOKEN' api_password = os.getenv("PROXY_TOKEN") url_main = os.getenv("PROXY_URL") # Check if the environment variable is set - If not, return an error if not api_password: raise HTTPException(status_code=500, detail="Server config error: TOKEN env variable is not set!") api_url = url_main # The external API URL params = {"password": api_password} try: # Try to Make the request to the external API response = requests.get(api_url, params=params, timeout=10) # type: ignore # 10-second timeout # Raise an exception for bad status codes response.raise_for_status() # Return the JSON data from the response return response.json() # Handle the Exception except requests.exceptions.RequestException as e: # Handle network errors, timeouts, or bad responses from the external API raise HTTPException(status_code=503, detail="Failed to fetch data from the upstream API!") if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=7860)