Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 26 additions & 14 deletions hub_main.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
from dotenv import load_dotenv

from fastapi import FastAPI, Request, HTTPException, status, Depends, BackgroundTasks
from fastapi import FastAPI, Request, HTTPException, status, Depends, BackgroundTasks, Response
from fastapi.responses import StreamingResponse
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
Expand Down Expand Up @@ -203,10 +204,13 @@ def register_user(payload: UserCreate, db: Session = Depends(get_user_db)):

return UserRegistrationResponse(**response_data)

def generate_jwt(email: str, private_key: str) -> dict:
def generate_jwt(email: str, private_key: str, role: str = "farmer") -> dict:
from datetime import timezone
expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
token = jwt.encode({"sub": email, "type": "access", "exp": expire}, private_key, algorithm=ALGORITHM, headers={"kid": "hub-key-1"})
claims = {"sub": email, "type": "access", "exp": expire}
if role == "authority":
claims["capabilities"] = ["trace-forward"]
token = jwt.encode(claims, private_key, algorithm=ALGORITHM, headers={"kid": "hub-key-1"})
return {"access_token": token, "token_type": "bearer"}

@app.post("/users/login", tags=["Authentication"])
Expand All @@ -219,7 +223,7 @@ def frontend_login(form_data: OAuth2PasswordRequestForm = Depends(), db: Session
raise HTTPException(status_code=403, detail="Account is not active.")

system_key = get_or_create_rsa_key(db)
return generate_jwt(user.email, system_key.private_key)
return generate_jwt(user.email, system_key.private_key, role=user.role)


class ApiKeyRequest(BaseModel):
Expand Down Expand Up @@ -264,7 +268,7 @@ def programmatic_token(payload: ClientCredentialPayload, db: Session = Depends(g
raise HTTPException(status_code=403, detail="Account is deactivated")

system_key = get_or_create_rsa_key(db)
return generate_jwt(user.email, system_key.private_key)
return generate_jwt(user.email, system_key.private_key, role=user.role)



Expand Down Expand Up @@ -347,12 +351,16 @@ async def calculate_cross_region_overlap(payload: OverlapRequest, request: Reque
headers["content-length"] = str(len(forward_body))

proxy_req = client.build_request("POST", target_url, headers=headers, content=forward_body)
proxy_resp = await client.send(proxy_req, stream=True)
proxy_resp = await client.send(proxy_req)

return StreamingResponse(
proxy_resp.aiter_raw(),
headers_to_return = dict(proxy_resp.headers)
headers_to_return.pop("content-encoding", None)
headers_to_return.pop("content-length", None)

return Response(
content=proxy_resp.content,
status_code=proxy_resp.status_code,
headers=proxy_resp.headers
headers=headers_to_return
)

print("[ORCHESTRATOR] Action: Fetching WKTs and calculating cross-border overlap using Shapely.")
Expand Down Expand Up @@ -457,7 +465,7 @@ async def route_to_regional_registry(endpoint_path: str, request: Request, db: S
if not geo_id and request.method in ["POST", "PUT"] and body_bytes:
try:
payload = json.loads(body_bytes.decode('utf-8'))
geo_id = payload.get("geo_id") or payload.get("geo_id_field_1")
geo_id = payload.get("geo_id") or payload.get("geo_id_field_1") or payload.get("seed_geoid")
except:
pass
if geo_id:
Expand All @@ -480,7 +488,7 @@ async def route_to_regional_registry(endpoint_path: str, request: Request, db: S
if query_params:
target_url = f"{target_url}?{query_params.decode('utf-8')}"

async with httpx.AsyncClient() as client:
async with httpx.AsyncClient(timeout=30.0) as client:
try:
headers = {k: v for k, v in request.headers.items() if k.lower() not in ("host",)}

Expand All @@ -490,12 +498,16 @@ async def route_to_regional_registry(endpoint_path: str, request: Request, db: S
headers=headers,
content=body_bytes
)
proxy_resp = await client.send(proxy_req, stream=True)
proxy_resp = await client.send(proxy_req)

headers_to_return = dict(proxy_resp.headers)
headers_to_return.pop("content-encoding", None)
headers_to_return.pop("content-length", None)

return StreamingResponse(
proxy_resp.aiter_raw(),
return Response(
content=proxy_resp.content,
status_code=proxy_resp.status_code,
headers=proxy_resp.headers
headers=headers_to_return
)
except httpx.RequestError as exc:
raise HTTPException(
Expand Down