feat: Asset Catalog system, PostGIS integration and RobotScout V1

backend/app/core/security.py

@@ -1,65 +1,48 @@
+import secrets
+import string
 from datetime import datetime, timedelta, timezone
-from typing import Optional, Dict, Any
+from typing import Optional, Dict, Any, Tuple
 import bcrypt
 from jose import jwt, JWTError
 from app.core.config import settings
+from fastapi_limiter import FastAPILimiter
+from fastapi_limiter.depends import RateLimiter
 
-# Master Book 5.0: RBAC Rank Definition Matrix
-# Ezek a szintek határozzák meg a hozzáférést a Middleware szintjén.
-RANK_MAP = {
-    "superadmin": 100,
-    "country_admin": 80,
-    "region_admin": 60,
-    "moderator": 40,
-    "sales": 20,
-    "user": 10,
-    "service": 15,
-    "fleet_manager": 25,
-    "driver": 5
+# Ezt az auth végpontokhoz adjuk hozzá:
+# @router.post("/login", dependencies=[Depends(RateLimiter(times=5, seconds=60))])
+
+DEFAULT_RANK_MAP = {
+    "superadmin": 100, "admin": 80, "fleet_manager": 25,
+    "service": 15, "user": 10, "driver": 5
 }
 
+def generate_secure_slug(length: int = 12) -> str:
+    """Biztonságos kód generálása (pl. mappákhoz)."""
+    alphabet = string.ascii_lowercase + string.digits
+    return ''.join(secrets.choice(alphabet) for _ in range(length))
+
 def verify_password(plain_password: str, hashed_password: str) -> bool:
-    """Összehasonlítja a sima szöveges jelszót a hash-elt változattal."""
-    if not hashed_password:
-        return False
+    if not hashed_password: return False
     try:
-        return bcrypt.checkpw(
-            plain_password.encode("utf-8"), 
-            hashed_password.encode("utf-8")
-        )
-    except Exception:
-        return False
+        return bcrypt.checkpw(plain_password.encode("utf-8"), hashed_password.encode("utf-8"))
+    except Exception: return False
 
 def get_password_hash(password: str) -> str:
-    """Létrehozza a jelszó hash-elt változatát."""
-    salt = bcrypt.gensalt()
-    return bcrypt.hashpw(password.encode("utf-8"), salt).decode("utf-8")
+    return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
 
-def create_access_token(data: Dict[str, Any], expires_delta: Optional[timedelta] = None) -> str:
-    """
-    Létrehozza a JWT access tokent bővített RBAC adatokkal.
-    Várt kulcsok: sub (user_id), role, rank, scope_level, scope_id
-    """
+def create_tokens(data: Dict[str, Any], access_delta: Optional[timedelta] = None, refresh_delta: Optional[timedelta] = None) -> Tuple[str, str]:
+    """Access és Refresh token generálása."""
     to_encode = data.copy()
-    if expires_delta:
-        expire = datetime.now(timezone.utc) + expires_delta
-    else:
-        expire = datetime.now(timezone.utc) + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
+    now = datetime.now(timezone.utc)
+    acc_min = access_delta if access_delta else timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
+    access_payload = {**to_encode, "exp": now + acc_min, "iat": now, "type": "access", "iss": "service-finder-auth"}
+    access_token = jwt.encode(access_payload, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
     
-    # Rendszer szintű metaadatok hozzáadása
-    to_encode.update({
-        "exp": expire,
-        "iat": datetime.now(timezone.utc),
-        "iss": "service-finder-auth"
-    })
-    
-    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
-    return encoded_jwt
+    ref_days = refresh_delta if refresh_delta else timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS)
+    refresh_payload = {"sub": str(to_encode.get("sub")), "exp": now + ref_days, "iat": now, "type": "refresh"}
+    refresh_token = jwt.encode(refresh_payload, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
+    return access_token, refresh_token
 
 def decode_token(token: str) -> Optional[Dict[str, Any]]:
-    """JWT token visszafejtése és validálása."""
-    try:
-        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
-        return payload
-    except JWTError:
-        return None
+    try: return jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
+    except JWTError: return None