111 lines
No EOL
3.4 KiB
Python
111 lines
No EOL
3.4 KiB
Python
import logging
|
|
import os
|
|
from logging.handlers import TimedRotatingFileHandler
|
|
from dotenv import load_dotenv
|
|
import secrets
|
|
from fastapi import FastAPI, Query, Depends, HTTPException, status, Request
|
|
from fastapi.security import APIKeyHeader
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from pydantic import BaseModel
|
|
from typing import List
|
|
|
|
from endpoint_compare import compare_string
|
|
|
|
load_dotenv()
|
|
|
|
os.makedirs("logs", exist_ok=True)
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s - %(levelname)s - %(message)s",
|
|
filename="logs/api_zugriffe.log",
|
|
filemode="a",
|
|
force=True
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
API_KEY_NAME = "X-API-Key"
|
|
api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=True)
|
|
|
|
app = FastAPI(title="SAP Endpoint Service")
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
class BulkCompareRequest(BaseModel):
|
|
endpoints: List[str]
|
|
|
|
def get_api_key(request: Request, api_key: str = Depends(api_key_header)):
|
|
client_ip = request.client.host if request.client else "Unbekannte IP"
|
|
method = request.method
|
|
path = request.url.path
|
|
|
|
if len(api_key) > 6:
|
|
masked_key = f"{api_key[:3]}...{api_key[-3:]}"
|
|
else:
|
|
masked_key = "***"
|
|
|
|
env_api_keys_str = os.getenv("APIKEYS", "").strip()
|
|
|
|
if not env_api_keys_str:
|
|
logger.error("Serverkonfigurationsfehler: Keine API-Keys in .env gefunden.")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Server configuration error: API Keys not set."
|
|
)
|
|
|
|
valid_keys = [k.strip() for k in env_api_keys_str.split(",") if k.strip()]
|
|
|
|
for valid_key in valid_keys:
|
|
if secrets.compare_digest(api_key, valid_key):
|
|
logger.info(f"Erfolgreicher Zugriff | IP: {client_ip} | Request: {method} {path} | Key: {masked_key}")
|
|
return api_key
|
|
|
|
logger.warning(f"Abgelehnter Zugriff (Falscher Key) | IP: {client_ip} | Request: {method} {path} | Key: {masked_key}")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid API Key"
|
|
)
|
|
|
|
@app.get("/")
|
|
def read_root(api_key: str = Depends(get_api_key)):
|
|
return {"status": "running", "message": "SAP Endpoint Compare API"}
|
|
|
|
@app.get("/informations/bulk")
|
|
def compare_multiple_endpoints(endpoints: List[str] = Query(..., description="Liste der zu prüfenden Endpunkte"),
|
|
api_key: str = Depends(get_api_key)
|
|
):
|
|
|
|
results = []
|
|
for ep in endpoints:
|
|
message, status, deprecated, decommissioned = compare_string(ep)
|
|
results.append({
|
|
"endpoint": ep,
|
|
"status": status,
|
|
"message": message,
|
|
"deprecated": deprecated,
|
|
"decommissioned": decommissioned
|
|
})
|
|
|
|
return {
|
|
"total_checked": len(results),
|
|
"results": results
|
|
}
|
|
|
|
@app.get("/informations/{endpoint:path}")
|
|
def read_item(endpoint: str, api_key: str = Depends(get_api_key)):
|
|
message, status, deprecated, decommissioned = compare_string(endpoint)
|
|
|
|
return {
|
|
"endpoint": endpoint,
|
|
"status": status,
|
|
"message": message,
|
|
"deprecated": deprecated,
|
|
"decommissioned": decommissioned
|
|
} |