- Auto-fixed 4,496 lint issues (import sorting, modern syntax, etc.) - Added ignore rules for patterns intentional in this codebase: E402 (late imports), E712 (SQLAlchemy filters), B904 (raise from), SIM108/SIM105/SIM117 (readability preferences) - Added per-file ignores for tests and scripts - Excluded broken scripts/rename_terminology.py (has curly quotes) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
196 lines
6.2 KiB
Python
196 lines
6.2 KiB
Python
# app/modules/tenancy/routes/api/store_auth.py
|
|
"""
|
|
Store team authentication endpoints.
|
|
|
|
Implements dual token storage with path restriction:
|
|
- Sets HTTP-only cookie with path=/store (restricted to store routes only)
|
|
- Returns token in response for localStorage (API calls)
|
|
|
|
This prevents:
|
|
- Store cookies from being sent to admin routes
|
|
- Admin cookies from being sent to store routes
|
|
- Cross-context authentication confusion
|
|
"""
|
|
|
|
import logging
|
|
|
|
from fastapi import APIRouter, Depends, Request, Response
|
|
from pydantic import BaseModel
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.api.deps import get_current_store_api
|
|
from app.core.database import get_db
|
|
from app.core.environment import should_use_secure_cookies
|
|
from app.modules.core.services.auth_service import auth_service
|
|
from app.modules.tenancy.exceptions import InvalidCredentialsException
|
|
from middleware.store_context import get_current_store
|
|
from models.schema.auth import LogoutResponse, StoreUserResponse, UserContext, UserLogin
|
|
|
|
store_auth_router = APIRouter(prefix="/auth")
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# Response model for store login
|
|
class StoreLoginResponse(BaseModel):
|
|
access_token: str
|
|
token_type: str
|
|
expires_in: int
|
|
user: dict
|
|
store: dict
|
|
store_role: str
|
|
|
|
|
|
@store_auth_router.post("/login", response_model=StoreLoginResponse)
|
|
def store_login(
|
|
user_credentials: UserLogin,
|
|
request: Request,
|
|
response: Response,
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""
|
|
Store team member login.
|
|
|
|
Authenticates users who are part of a store team.
|
|
Validates against store context if available.
|
|
|
|
Sets token in two places:
|
|
1. HTTP-only cookie with path=/store (for browser page navigation)
|
|
2. Response body (for localStorage and API calls)
|
|
|
|
Prevents admin users from logging into store portal.
|
|
"""
|
|
# Try to get store from middleware first
|
|
store = get_current_store(request)
|
|
|
|
# If no store from middleware, try to get from request body
|
|
if not store and hasattr(user_credentials, "store_code"):
|
|
store_code = getattr(user_credentials, "store_code", None)
|
|
if store_code:
|
|
store = auth_service.get_store_by_code(db, store_code)
|
|
|
|
# Authenticate user
|
|
login_result = auth_service.login_user(db=db, user_credentials=user_credentials)
|
|
user = login_result["user"]
|
|
|
|
# CRITICAL: Prevent admin users from using store login
|
|
if user.role == "admin":
|
|
logger.warning(f"Admin user attempted store login: {user.username}")
|
|
raise InvalidCredentialsException(
|
|
"Admins cannot access store portal. Please use admin portal."
|
|
)
|
|
|
|
# Determine store and role
|
|
store_role = "Member"
|
|
|
|
if store:
|
|
# Check if user has access to this store
|
|
has_access, role = auth_service.get_user_store_role(db, user, store)
|
|
|
|
if has_access:
|
|
store_role = role
|
|
else:
|
|
logger.warning(
|
|
f"User {user.username} attempted login to store {store.store_code} "
|
|
f"but is not authorized"
|
|
)
|
|
raise InvalidCredentialsException("You do not have access to this store")
|
|
else:
|
|
# No store context - find which store this user belongs to
|
|
store, store_role = auth_service.find_user_store(user)
|
|
|
|
if not store:
|
|
raise InvalidCredentialsException("User is not associated with any store")
|
|
|
|
logger.info(
|
|
f"Store team login successful: {user.username} "
|
|
f"for store {store.store_code} as {store_role}"
|
|
)
|
|
|
|
# Create store-scoped access token with store information
|
|
token_data = auth_service.auth_manager.create_access_token(
|
|
user=user,
|
|
store_id=store.id,
|
|
store_code=store.store_code,
|
|
store_role=store_role,
|
|
)
|
|
|
|
# Set HTTP-only cookie for browser navigation
|
|
# CRITICAL: path=/store restricts cookie to store routes only
|
|
response.set_cookie(
|
|
key="store_token",
|
|
value=token_data["access_token"],
|
|
httponly=True, # JavaScript cannot access (XSS protection)
|
|
secure=should_use_secure_cookies(), # HTTPS only in production/staging
|
|
samesite="lax", # CSRF protection
|
|
max_age=token_data["expires_in"], # Match JWT expiry
|
|
path="/store", # RESTRICTED TO STORE ROUTES ONLY
|
|
)
|
|
|
|
logger.debug(
|
|
f"Set store_token cookie with {token_data['expires_in']}s expiry "
|
|
f"(path=/store, httponly=True, secure={should_use_secure_cookies()})"
|
|
)
|
|
|
|
# Return full login response with store-scoped token
|
|
return StoreLoginResponse(
|
|
access_token=token_data["access_token"],
|
|
token_type=token_data["token_type"],
|
|
expires_in=token_data["expires_in"],
|
|
user={
|
|
"id": user.id,
|
|
"username": user.username,
|
|
"email": user.email,
|
|
"role": user.role,
|
|
"is_active": user.is_active,
|
|
},
|
|
store={
|
|
"id": store.id,
|
|
"store_code": store.store_code,
|
|
"subdomain": store.subdomain,
|
|
"name": store.name,
|
|
"is_active": store.is_active,
|
|
"is_verified": store.is_verified,
|
|
},
|
|
store_role=store_role,
|
|
)
|
|
|
|
|
|
@store_auth_router.post("/logout", response_model=LogoutResponse)
|
|
def store_logout(response: Response):
|
|
"""
|
|
Store team member logout.
|
|
|
|
Clears the store_token cookie.
|
|
Client should also remove token from localStorage.
|
|
"""
|
|
logger.info("Store logout")
|
|
|
|
# Clear the cookie (must match path used when setting)
|
|
response.delete_cookie(
|
|
key="store_token",
|
|
path="/store",
|
|
)
|
|
|
|
logger.debug("Deleted store_token cookie")
|
|
|
|
return LogoutResponse(message="Logged out successfully")
|
|
|
|
|
|
@store_auth_router.get("/me", response_model=StoreUserResponse)
|
|
def get_current_store_user(
|
|
user: UserContext = Depends(get_current_store_api), db: Session = Depends(get_db)
|
|
):
|
|
"""
|
|
Get current authenticated store user.
|
|
|
|
This endpoint can be called to verify authentication and get user info.
|
|
Requires Authorization header (header-only authentication for API endpoints).
|
|
"""
|
|
return StoreUserResponse(
|
|
id=user.id,
|
|
username=user.username,
|
|
email=user.email,
|
|
role=user.role,
|
|
is_active=user.is_active,
|
|
)
|