""" Authentication & Authorization Service Provides password hashing, JWT token management, and permission checking. """ from datetime import datetime, timedelta from typing import Optional import hashlib import hmac import os # Try to import passlib, fallback to simple hashing from ...utils.timezone import now_cn_aware, now_cn_naive try: from passlib.context import CryptContext from jose import JWTError, jwt HAS_PASSLIB = True except ImportError: HAS_PASSLIB = False # Fallback JWT implementation try: import jwt as jwt_lib jwt = jwt_lib JWTError = Exception except ImportError: jwt = None JWTError = Exception # JWT Configuration SECRET_KEY = os.getenv("SECRET_KEY", "neural-sync-secret-key-change-in-production") ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_HOURS = 24 MAX_BCRYPT_PASSWORD_BYTES = 72 # Password hashing context (if passlib available) if HAS_PASSLIB: pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") class AuthService: """Service for authentication operations""" @staticmethod def password_utf8_bytes(password: str) -> int: return len(str(password or "").encode("utf-8")) @staticmethod def validate_password(password: str) -> None: if len(str(password or "")) < 6: raise ValueError("密码至少需要 6 个字符") if AuthService.password_utf8_bytes(password) > MAX_BCRYPT_PASSWORD_BYTES: raise ValueError("密码最长支持 72 个 UTF-8 字节，请缩短后重试") @staticmethod def verify_password(plain_password: str, hashed_password: str) -> bool: """Verify a plain password against a hashed password""" if HAS_PASSLIB: try: # 优先走 bcrypt（新数据） if pwd_context.verify(plain_password, hashed_password): return True except Exception: # 旧库可能是历史 SHA256 哈希，回退兼容 pass return hmac.compare_digest( AuthService._simple_hash(plain_password), hashed_password, ) # Simple SHA256 fallback for development return hmac.compare_digest( AuthService._simple_hash(plain_password), hashed_password ) @staticmethod def hash_password(password: str) -> str: """Hash a password for storage""" AuthService.validate_password(password) if HAS_PASSLIB: try: return pwd_context.hash(password) except ValueError as exc: if "72 bytes" in str(exc).lower() and AuthService.password_utf8_bytes(password) > MAX_BCRYPT_PASSWORD_BYTES: raise ValueError("密码最长支持 72 个 UTF-8 字节，请缩短后重试") from exc return AuthService._simple_hash(password) except Exception: return AuthService._simple_hash(password) else: # Simple SHA256 fallback for development return AuthService._simple_hash(password) @staticmethod def _simple_hash(password: str) -> str: """Simple SHA256 hash with salt for development use""" salt = SECRET_KEY[:16].encode() if SECRET_KEY else b'neural-sync-salt' return hashlib.sha256(password.encode() + salt).hexdigest() @staticmethod def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str: """Create a JWT access token""" if jwt is None: # Fallback: return a simple token (not recommended for production) import base64 import json token_data = { **data, "exp": (now_cn_naive() + (expires_delta or timedelta(hours=ACCESS_TOKEN_EXPIRE_HOURS))).timestamp() } return base64.b64encode(json.dumps(token_data).encode()).decode() to_encode = data.copy() if expires_delta: expire = now_cn_naive() + expires_delta else: expire = now_cn_naive() + timedelta(hours=ACCESS_TOKEN_EXPIRE_HOURS) to_encode.update({"exp": expire}) return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) @staticmethod def verify_token(token: str) -> Optional[str]: """ Verify a JWT token and return the username Returns None if token is invalid or expired """ if jwt is None: # Fallback: decode simple token import base64 import json try: token_data = json.loads(base64.b64decode(token).decode()) if token_data.get("exp", 0) > now_cn_aware().timestamp(): return token_data.get("sub") return None except: return None try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) username: str = payload.get("sub") return username except JWTError: return None MANAGER_ROLES = {"OWNER", "REVIEWER"} VALID_ROLE_TOKENS = {"OWNER", "BUILDER", "OPERATOR", "REVIEWER"} def canonicalize_role_token(value: str | None) -> str: token = str(value or "").strip().upper() return token if token in VALID_ROLE_TOKENS else token # Permission matrix for role-based access control PERMISSION_MATRIX = { "OWNER": [ "view_all_signals", "assign_tasks", "manage_users", "delete_users", "manage_domains", "view_projects", "view_context", "view_settings", "manage_staff", "manage_global_integrations", "approve_release_gate", "override_compliance_gate", ], "REVIEWER": [ "view_all_signals", "assign_tasks", "manage_domains", "view_projects", "view_context", "view_settings", "manage_staff", "manage_global_integrations", "approve_release_gate", ], "BUILDER": [ "view_assigned_signals", "create_experiments", "view_context", "work_on_dev_tasks", "use_prompt_packages", "sync_repo_progress", ], "OPERATOR": [ "view_assigned_signals", "view_assigned_projects", "ingest_feedback", "record_revenue_events", "view_operating_metrics", ], } def has_permission(user_role: str, permission: str) -> bool: """Check if a role has a specific permission""" roles = parse_role_values(user_role) return any(permission in PERMISSION_MATRIX.get(role, []) for role in roles) def has_any_role(user_role: str, allowed_roles: list) -> bool: """Check if user_role is in the list of allowed roles""" allowed = {canonicalize_role_token(str(role)) for role in allowed_roles} return any(role in allowed for role in parse_role_values(user_role)) def parse_role_values(user_role: str | None) -> list[str]: """Parse single/multi role string into uppercase role tokens.""" raw = str(user_role or "") if not raw: return [] values: list[str] = [] for token in raw.split(","): role = canonicalize_role_token(token) if role and role not in values: values.append(role) return values def can_manage_users(user_role: str) -> bool: """Check if user can manage other users (OWNER only)""" return has_any_role(user_role, ["OWNER"]) def can_assign_tasks(user_role: str) -> bool: """Check if user can assign tasks to others""" return has_any_role(user_role, ["OWNER", "REVIEWER"]) def can_view_ai_policy(user_role: str) -> bool: """AI policy pages are visible to governance roles only.""" return has_any_role(user_role, ["OWNER", "REVIEWER"]) def can_operate_ai_routing(user_role: str) -> bool: """AI routing diagnostics/tests are available to reviewer-level operators.""" return has_any_role(user_role, ["OWNER", "REVIEWER"]) def can_approve_ai_routing_change(user_role: str) -> bool: """High-risk AI routing/policy changes require owner approval in v1.""" return has_any_role(user_role, ["OWNER"])