98 lines
3.5 KiB
Python
98 lines
3.5 KiB
Python
from pydantic import BaseModel, field_serializer, field_validator, Field
|
|
from datetime import datetime
|
|
from typing import Optional
|
|
import re
|
|
import html
|
|
|
|
# Message schemas
|
|
class MessageBase(BaseModel):
|
|
message: str = Field(..., min_length=1, max_length=1000, description="Message content")
|
|
sender: str = Field(..., min_length=1, max_length=255, description="Sender name")
|
|
item_id: Optional[int] = Field(None, ge=0, description="Item ID")
|
|
amount: Optional[int] = Field(None, ge=0, description="Item amount")
|
|
|
|
# No validators on base class - they will be added to Create classes only
|
|
|
|
class MessageCreate(MessageBase):
|
|
@field_validator('message', 'sender')
|
|
@classmethod
|
|
def sanitize_text(cls, v):
|
|
"""Sanitize text input to prevent XSS and other attacks"""
|
|
if not v:
|
|
return v
|
|
# HTML escape to prevent XSS
|
|
v = html.escape(v.strip())
|
|
# Remove any remaining script tags or dangerous patterns
|
|
v = re.sub(r'<script.*?</script>', '', v, flags=re.IGNORECASE | re.DOTALL)
|
|
v = re.sub(r'javascript:', '', v, flags=re.IGNORECASE)
|
|
return v
|
|
|
|
@field_validator('sender')
|
|
@classmethod
|
|
def validate_sender(cls, v):
|
|
"""Validate sender name format"""
|
|
if not v:
|
|
return v
|
|
# Allow only alphanumeric, spaces, and common characters
|
|
if not re.match(r'^[a-zA-Z0-9\s\-_\.]+$', v):
|
|
raise ValueError('Sender name contains invalid characters')
|
|
return v
|
|
|
|
class Message(MessageBase):
|
|
id: int
|
|
created_at: datetime
|
|
|
|
@field_serializer('created_at')
|
|
def serialize_created_at(self, value: datetime) -> str:
|
|
"""Format datetime to match plugin expectations: 'u-M-d H:m:s' (no leading zeros)"""
|
|
return value.strftime("%Y-%-m-%-d %-H:%M:%S")
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
# Transaction schemas
|
|
class TransactionBase(BaseModel):
|
|
item_id: int = Field(..., ge=0, description="Item ID")
|
|
item: str = Field(..., min_length=1, max_length=255, description="Item name")
|
|
user: str = Field(..., min_length=1, max_length=255, description="User name")
|
|
amount: int = Field(..., description="Transaction amount")
|
|
|
|
# No validators on base class - they will be added to Create classes only
|
|
|
|
class TransactionCreate(TransactionBase):
|
|
@field_validator('item', 'user')
|
|
@classmethod
|
|
def sanitize_text(cls, v):
|
|
"""Sanitize text input to prevent XSS and other attacks"""
|
|
if not v:
|
|
return v
|
|
# HTML escape to prevent XSS
|
|
v = html.escape(v.strip())
|
|
# Remove any remaining script tags or dangerous patterns
|
|
v = re.sub(r'<script.*?</script>', '', v, flags=re.IGNORECASE | re.DOTALL)
|
|
v = re.sub(r'javascript:', '', v, flags=re.IGNORECASE)
|
|
return v
|
|
|
|
@field_validator('user')
|
|
@classmethod
|
|
def validate_user(cls, v):
|
|
"""Validate user name format"""
|
|
if not v:
|
|
return v
|
|
# Allow only alphanumeric, spaces, and common characters
|
|
if not re.match(r'^[a-zA-Z0-9\s\-_\.]+$', v):
|
|
raise ValueError('User name contains invalid characters')
|
|
return v
|
|
|
|
class Transaction(TransactionBase):
|
|
id: int
|
|
created_at: datetime
|
|
|
|
@field_serializer('created_at')
|
|
def serialize_created_at(self, value: datetime) -> str:
|
|
"""Format datetime to match plugin expectations: 'u-M-d H:m:s' (no leading zeros)"""
|
|
return value.strftime("%Y-%-m-%-d %-H:%M:%S")
|
|
|
|
class Config:
|
|
from_attributes = True
|