808fbf5c7c
<message>Update the _set_golden_from_path function to improve the handling of existing golden image files. Replace the existing unlink logic with a more robust method that safely removes files or broken symlinks using the missing_ok parameter. This change enhances the reliability of the backup upload process by ensuring that stale references are properly cleared before setting a new golden image path.
212 lines
7.3 KiB
Python
212 lines
7.3 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
SQLAlchemy and Pydantic models for GNSS Guard Server
|
|
"""
|
|
|
|
from datetime import datetime
|
|
from typing import Dict, Any, List, Optional
|
|
from sqlalchemy import Column, Integer, String, Float, Boolean, DateTime, ForeignKey, Text, Index
|
|
from sqlalchemy.orm import relationship, declarative_base
|
|
from pydantic import BaseModel, Field
|
|
import hashlib
|
|
import secrets
|
|
|
|
Base = declarative_base()
|
|
|
|
|
|
# =============================================================================
|
|
# SQLAlchemy Database Models
|
|
# =============================================================================
|
|
|
|
class Asset(Base):
|
|
"""Asset (client device) registered with the server"""
|
|
__tablename__ = "assets"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
name = Column(String(255), unique=True, nullable=False, index=True)
|
|
token_hash = Column(String(64), nullable=False) # SHA-256 hash of token
|
|
created_at = Column(DateTime, default=datetime.utcnow)
|
|
is_active = Column(Boolean, default=True)
|
|
description = Column(String(500), nullable=True)
|
|
|
|
# Telegram notification settings (optional override for this asset)
|
|
telegram_chat_id = Column(String(100), nullable=True) # Override default chat ID
|
|
telegram_enabled = Column(Boolean, default=True) # Enable/disable notifications for this asset
|
|
|
|
# Relationship to validation history
|
|
validations = relationship("ValidationHistory", back_populates="asset", cascade="all, delete-orphan")
|
|
|
|
# Relationship to notification state
|
|
notification_state = relationship("AssetNotificationState", back_populates="asset", uselist=False, cascade="all, delete-orphan")
|
|
|
|
@staticmethod
|
|
def hash_token(token: str) -> str:
|
|
"""Hash a token using SHA-256"""
|
|
return hashlib.sha256(token.encode()).hexdigest()
|
|
|
|
@staticmethod
|
|
def generate_token() -> str:
|
|
"""Generate a secure random token"""
|
|
return secrets.token_urlsafe(32)
|
|
|
|
def verify_token(self, token: str) -> bool:
|
|
"""Verify if provided token matches stored hash"""
|
|
return self.token_hash == self.hash_token(token)
|
|
|
|
|
|
class AssetNotificationState(Base):
|
|
"""Tracks the previous notification state for each asset to detect changes"""
|
|
__tablename__ = "asset_notification_state"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
asset_id = Column(Integer, ForeignKey("assets.id", ondelete="CASCADE"), unique=True, nullable=False)
|
|
|
|
# Previous state (JSON arrays stored as text)
|
|
prev_sources_missing = Column(Text, nullable=True) # JSON array
|
|
prev_sources_stale = Column(Text, nullable=True) # JSON array
|
|
prev_threshold_breached = Column(Boolean, default=False)
|
|
|
|
# Last notification timestamp
|
|
last_notification_at = Column(DateTime, nullable=True)
|
|
|
|
# Asset online/offline tracking
|
|
is_online = Column(Boolean, default=True) # Whether asset is currently reporting
|
|
last_validation_at = Column(DateTime, nullable=True) # Last time we received validation data
|
|
|
|
# Relationship
|
|
asset = relationship("Asset", back_populates="notification_state")
|
|
|
|
|
|
class ValidationHistory(Base):
|
|
"""Historical validation records from assets"""
|
|
__tablename__ = "validation_history"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
asset_id = Column(Integer, ForeignKey("assets.id", ondelete="CASCADE"), nullable=False)
|
|
|
|
# Validation timestamps
|
|
validation_timestamp = Column(String(50), nullable=False) # ISO format
|
|
validation_timestamp_unix = Column(Float, nullable=False, index=True)
|
|
|
|
# Validation result
|
|
is_valid = Column(Boolean, nullable=False)
|
|
|
|
# JSON fields stored as text
|
|
sources_missing = Column(Text, nullable=True) # JSON array
|
|
sources_stale = Column(Text, nullable=True) # JSON array
|
|
coordinate_differences = Column(Text, nullable=True) # JSON object
|
|
source_coordinates = Column(Text, nullable=True) # JSON object
|
|
validation_details = Column(Text, nullable=True) # JSON object
|
|
|
|
# Server-side metadata
|
|
received_at = Column(DateTime, default=datetime.utcnow, index=True)
|
|
|
|
# Relationship
|
|
asset = relationship("Asset", back_populates="validations")
|
|
|
|
# Indexes for common queries
|
|
__table_args__ = (
|
|
Index('ix_validation_asset_timestamp', 'asset_id', 'validation_timestamp_unix'),
|
|
)
|
|
|
|
|
|
# =============================================================================
|
|
# Pydantic Request/Response Models
|
|
# =============================================================================
|
|
|
|
class AssetCreate(BaseModel):
|
|
"""Request model for creating a new asset"""
|
|
name: str = Field(..., min_length=1, max_length=255)
|
|
description: Optional[str] = Field(None, max_length=500)
|
|
telegram_chat_id: Optional[str] = Field(None, max_length=100) # Override default chat ID
|
|
telegram_enabled: bool = True # Enable notifications for this asset
|
|
|
|
|
|
class AssetResponse(BaseModel):
|
|
"""Response model for asset data"""
|
|
id: int
|
|
name: str
|
|
is_active: bool
|
|
created_at: datetime
|
|
description: Optional[str] = None
|
|
telegram_chat_id: Optional[str] = None
|
|
telegram_enabled: bool = True
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class AssetWithToken(AssetResponse):
|
|
"""Response model for newly created asset (includes token)"""
|
|
token: str # Only returned when asset is created
|
|
|
|
|
|
class AssetImport(BaseModel):
|
|
"""Request model for importing an asset with a specific token"""
|
|
name: str = Field(..., min_length=1, max_length=255)
|
|
token: str = Field(..., min_length=32, max_length=128)
|
|
description: Optional[str] = Field(None, max_length=500)
|
|
telegram_chat_id: Optional[str] = Field(None, max_length=100)
|
|
telegram_enabled: bool = True
|
|
|
|
|
|
class AssetBatchImport(BaseModel):
|
|
"""Request model for batch importing assets"""
|
|
assets: List[AssetImport]
|
|
|
|
|
|
class ValidationSubmission(BaseModel):
|
|
"""Request model for submitting validation data"""
|
|
validation_timestamp: str
|
|
validation_timestamp_unix: float
|
|
is_valid: bool
|
|
sources_missing: List[str] = []
|
|
sources_stale: List[str] = []
|
|
coordinate_differences: Dict[str, Any] = {}
|
|
source_coordinates: Dict[str, Any] = {}
|
|
validation_details: Dict[str, Any] = {}
|
|
|
|
|
|
class ValidationBatchSubmission(BaseModel):
|
|
"""Request model for submitting multiple validation records"""
|
|
records: List[ValidationSubmission]
|
|
|
|
|
|
class ValidationResponse(BaseModel):
|
|
"""Response model for validation data"""
|
|
id: int
|
|
asset_name: str
|
|
validation_timestamp: str
|
|
validation_timestamp_unix: float
|
|
is_valid: bool
|
|
sources_missing: List[str]
|
|
sources_stale: List[str]
|
|
coordinate_differences: Dict[str, Any]
|
|
source_coordinates: Dict[str, Any]
|
|
validation_details: Dict[str, Any]
|
|
received_at: datetime
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class AssetStatus(BaseModel):
|
|
"""Current status of an asset (latest validation)"""
|
|
asset_name: str
|
|
is_online: bool # Has reported in last 5 minutes
|
|
last_seen: Optional[datetime] = None
|
|
latest_validation: Optional[ValidationResponse] = None
|
|
|
|
|
|
class LoginRequest(BaseModel):
|
|
"""Request model for user login"""
|
|
username: str
|
|
password: str
|
|
|
|
|
|
class LoginResponse(BaseModel):
|
|
"""Response model for successful login"""
|
|
message: str
|
|
username: str
|
|
|