This commit is contained in:
HeyItsChloe
2026-03-20 09:37:00 -07:00
parent 9adcaa20c7
commit 75ddf8f737

View File

@@ -36,7 +36,7 @@ def _sanitize_text(text: str) -> str:
Sanitized text safe for storage and display
"""
# Remove control characters (except newlines and tabs which are valid in messages)
sanitized = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]", "", text)
sanitized = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]', '', text)
# Escape HTML to prevent XSS if displayed in admin tools
sanitized = html.escape(sanitized)
return sanitized
@@ -46,9 +46,9 @@ class FormSubmissionRequest(BaseModel):
"""Request model for form submission."""
form_type: str = Field(
..., max_length=50, description="Type of form being submitted"
..., max_length=50, description='Type of form being submitted'
)
answers: dict[str, Any] = Field(..., description="Form answers as key-value pairs")
answers: dict[str, Any] = Field(..., description='Form answers as key-value pairs')
class FormSubmissionResponse(BaseModel):
@@ -62,13 +62,13 @@ class FormSubmissionResponse(BaseModel):
class EnterpriseLeadAnswers(BaseModel):
"""Validation model for enterprise lead form answers."""
request_type: str = Field(..., pattern="^(saas|self-hosted)$")
request_type: str = Field(..., pattern='^(saas|self-hosted)$')
name: str = Field(..., min_length=1, max_length=255)
company: str = Field(..., min_length=1, max_length=255)
email: EmailStr = Field(..., max_length=255)
message: str = Field(..., min_length=1, max_length=2000)
@field_validator("name", "company", "message", mode="after")
@field_validator('name', 'company', 'message', mode='after')
@classmethod
def sanitize_text_fields(cls, v: str) -> str:
"""Sanitize text fields to prevent XSS and remove control characters."""
@@ -91,7 +91,7 @@ def _get_user_id_from_request(request: Request) -> UUID | None:
InvalidUserIdError: If user_id exists but is not a valid UUID,
indicating a bug in the auth system.
"""
user_auth: UserAuth | None = getattr(request.state, "user_auth", None)
user_auth: UserAuth | None = getattr(request.state, 'user_auth', None)
if user_auth is None:
return None
@@ -103,9 +103,9 @@ def _get_user_id_from_request(request: Request) -> UUID | None:
except ValueError as e:
# This should never happen if auth is working correctly.
# Fail fast to surface the bug instead of silently masking it.
logger.error(f"Auth system returned invalid user_id format: {user_id}")
logger.error(f'Auth system returned invalid user_id format: {user_id}')
raise InvalidUserIdError(
f"Invalid user_id format from auth: {user_id}"
f'Invalid user_id format from auth: {user_id}'
) from e
return None
@@ -117,11 +117,11 @@ def _validate_enterprise_lead_answers(answers: dict[str, Any]) -> None:
except ValidationError as e:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Invalid enterprise lead form answers: {str(e)}",
detail=f'Invalid enterprise lead form answers: {str(e)}',
)
@router.post("/submit", status_code=status.HTTP_201_CREATED)
@router.post('/submit', status_code=status.HTTP_201_CREATED)
async def submit_form(
request: Request,
submission: FormSubmissionRequest,
@@ -144,7 +144,7 @@ async def submit_form(
await form_submit_rate_limiter.hit('form_submit', client_ip)
# Validate form type
valid_form_types = {"enterprise_lead"}
valid_form_types = {'enterprise_lead'}
if submission.form_type not in valid_form_types:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
@@ -152,7 +152,7 @@ async def submit_form(
)
# Validate and sanitize answers based on form type
if submission.form_type == "enterprise_lead":
if submission.form_type == 'enterprise_lead':
_validate_enterprise_lead_answers(submission.answers)
# Get user ID if authenticated (optional)
@@ -163,7 +163,7 @@ async def submit_form(
# Return 500 to surface auth system bug
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Internal authentication error",
detail='Internal authentication error',
)
# Create submission record
@@ -172,7 +172,7 @@ async def submit_form(
id=submission_id,
form_type=submission.form_type,
answers=submission.answers,
status="pending",
status='pending',
user_id=user_id,
)
@@ -183,11 +183,11 @@ async def submit_form(
await session.refresh(new_submission)
logger.info(
"form_submission_created",
'form_submission_created',
extra={
"submission_id": str(submission_id),
"form_type": submission.form_type,
"user_id": str(user_id) if user_id else None,
'submission_id': str(submission_id),
'form_type': submission.form_type,
'user_id': str(user_id) if user_id else None,
},
)