Spaces:

dn-scribe
/

attstore

Runtime error

attstore / models.py
fix: ux issues regarding admin vs regular user behavior (not allowing same product name accorss users, and some ux issuse)
7438530 11 months ago
Raw History Blame Contribute Delete
28.7 kB
from app import db
from sqlalchemy import func
from datetime import datetime
from sqlalchemy import JSON, Text, Boolean, Integer, String, DateTime, ForeignKey, UniqueConstraint
from sqlalchemy.orm import relationship
import secrets
from werkzeug.security import generate_password_hash, check_password_hash
from sqlalchemy.ext.mutable import MutableDict


def _default_attribute_mapping():
    return {
        'email': 'email',
        'username': 'username',
        'first_name': 'firstName',
        'last_name': 'lastName'
    }


class Tenant(db.Model):
    __tablename__ = 'tenants'

    id = db.Column(Integer, primary_key=True)
    name = db.Column(String(128), nullable=False)
    slug = db.Column(String(64), unique=True, nullable=False)
    description = db.Column(Text, nullable=True)
    active = db.Column(Boolean, default=True)
    allow_password_auth = db.Column(Boolean, default=True)
    created_at = db.Column(DateTime, default=datetime.utcnow)
    updated_at = db.Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    __table_args__ = (
        db.Index('ix_tenants_slug', 'slug'),
        db.Index('ix_tenants_active', 'active'),
    )

    def to_dict(self):
        return {
            'id': self.id,
            'name': self.name,
            'slug': self.slug,
            'description': self.description,
            'active': self.active,
            'allow_password_auth': self.allow_password_auth,
            'created_at': self.created_at.isoformat() if self.created_at else None,
            'updated_at': self.updated_at.isoformat() if self.updated_at else None,
        }


class Badge(db.Model):
    __tablename__ = 'badges'
    id = db.Column(Integer, primary_key=True)
    name = db.Column(String(64), unique=True, nullable=False)
    description = db.Column(Text, nullable=True)
    owner_id = db.Column(Integer, ForeignKey('users.id'), nullable=True)  # Null for global/admin badges
    is_global = db.Column(Boolean, default=False)
    last_modified = db.Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    products = db.Column(String(255), nullable=True)  # Product name pattern for LIKE query
    criteria = db.Column(Text, nullable=True)  # SQL WHERE clause after user_id/product_key
    table = db.Column(String(64), default='policy_results')  # Table to query (attestations or policy_results)
    positive = db.Column(Boolean, default=False)  # If true, 0 is bad (red), >0 is good (green)
    visible = db.Column(Boolean, default=True)  # New field: badge visibility

    owner = relationship('User', backref='badges', foreign_keys=[owner_id])

    def to_dict(self):
        return {
            'id': self.id,
            'name': self.name,
            'description': self.description,
            'owner_id': self.owner_id,
            'is_global': self.is_global,
            'last_modified': self.last_modified.isoformat() if self.last_modified else None,
            'products': self.products,
            'criteria': self.criteria,
            'table': self.table,
            'positive': self.positive,
            'visible': self.visible
        }
from sqlalchemy import func
from datetime import datetime, timedelta
import secrets
from sqlalchemy import JSON, Text, Boolean, Integer, String, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from werkzeug.security import generate_password_hash, check_password_hash

class User(db.Model):
    __tablename__ = 'users'
    
    id = db.Column(Integer, primary_key=True)
    username = db.Column(String(64), unique=True, nullable=False)
    email = db.Column(String(120), nullable=False)
    password_hash = db.Column(String(256), nullable=False)
    created_at = db.Column(DateTime, default=datetime.utcnow)
    is_active = db.Column(Boolean, default=True)
    is_admin = db.Column(Boolean, default=False)
    password_change_required = db.Column(Boolean, default=False)
    last_password_change = db.Column(DateTime, nullable=True)
    failed_login_attempts = db.Column(Integer, default=0)
    last_failed_login = db.Column(DateTime, nullable=True)
    last_successful_login = db.Column(DateTime, nullable=True)
    tenant_id = db.Column(Integer, ForeignKey('tenants.id'), nullable=True)
    
    tenant = relationship('Tenant', backref='users')

    def set_password(self, password):
        self.password_hash = generate_password_hash(password)
        self.last_password_change = datetime.utcnow()
    
    def check_password(self, password):
        return check_password_hash(self.password_hash, password)
    
    def record_login_success(self):
        """Record a successful login attempt"""
        self.failed_login_attempts = 0
        self.last_successful_login = datetime.utcnow()
    
    def record_login_failure(self):
        """Record a failed login attempt"""
        self.failed_login_attempts += 1
        self.last_failed_login = datetime.utcnow()
        
        # Only deactivate non-admin accounts after 50 consecutive failures
        if self.username != 'admin' and self.failed_login_attempts >= 50:
            self.is_active = False
            return True  # Account was deactivated
        return False  # Account was not deactivated
    
    __table_args__ = (
        db.Index('ix_users_username', 'username'),
        db.Index('ix_users_is_active', 'is_active'),
        db.Index('ix_users_is_admin', 'is_admin'),
        db.Index('ix_users_tenant_id', 'tenant_id'),
    )

    def to_dict(self):
        return {
            'id': self.id,
            'username': self.username,
            'email': self.email,
            'created_at': self.created_at.isoformat() if self.created_at else None,
            'is_active': self.is_active,
            'is_admin': self.is_admin,
            'password_change_required': self.password_change_required,
            'last_password_change': self.last_password_change.isoformat() if self.last_password_change else None,
            'failed_login_attempts': self.failed_login_attempts,
            'last_failed_login': self.last_failed_login.isoformat() if self.last_failed_login else None,
            'last_successful_login': self.last_successful_login.isoformat() if self.last_successful_login else None,
            'tenant_id': self.tenant_id,
        }

class ApiKey(db.Model):
    __tablename__ = 'api_keys'
    
    id = db.Column(Integer, primary_key=True)
    key = db.Column(String(255), unique=True, nullable=False)
    name = db.Column(String(255))  # Optional description/name for the key
    user_id = db.Column(Integer, ForeignKey('users.id'), nullable=False)
    created_by = db.Column(Integer, ForeignKey('users.id'), nullable=False)
    expiration_date = db.Column(DateTime, nullable=True)  # NULL means no expiration
    created_at = db.Column(DateTime, default=datetime.utcnow)
    updated_at = db.Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    is_active = db.Column(Boolean, default=True)
    
    # Relationships
    user = relationship("User", foreign_keys=[user_id], backref="api_keys")
    creator = relationship("User", foreign_keys=[created_by])
    
    @staticmethod
    def generate_key(username):
        """Generate a secure API key in base64 format: username:secret"""
        import base64
        secret = secrets.token_urlsafe(32)
        credentials = f"{username}:{secret}"
        return base64.b64encode(credentials.encode()).decode()
    
    @staticmethod
    def create_key(user_id, created_by, name=None, expiration_days=365):
        """Create a new API key with optional expiration"""
        from app import db
        # Get the user to extract username for key generation
        user = db.session.get(User, user_id)
        if not user:
            raise ValueError("User not found")
            
        key = ApiKey.generate_key(user.username)
        expiration_date = None
        
        if expiration_days is not None:
            expiration_date = datetime.utcnow() + timedelta(days=expiration_days)
        
        api_key = ApiKey()
        api_key.key = key
        api_key.name = name
        api_key.user_id = user_id
        api_key.created_by = created_by
        api_key.expiration_date = expiration_date
        
        return api_key
    
    def is_expired(self):
        """Check if the API key has expired"""
        if self.expiration_date is None:
            return False
        return datetime.utcnow() > self.expiration_date
    
    def is_valid(self):
        """Check if the API key is valid (active and not expired)"""
        return self.is_active and not self.is_expired()
    
    def get_client_credentials(self):
        """Extract client_id and client_secret from base64 encoded key"""
        try:
            import base64
            decoded = base64.b64decode(self.key.encode()).decode()
            if ':' in decoded:
                client_id, client_secret = decoded.split(':', 1)
                return client_id, client_secret
        except:
            pass
        return None, None
    
    __table_args__ = (
        db.Index('ix_api_keys_key', 'key'),
        db.Index('ix_api_keys_user_id', 'user_id'),
        db.Index('ix_api_keys_is_active', 'is_active'),
        db.Index('ix_api_keys_expiration_date', 'expiration_date'),
    )

    def to_dict(self):
        client_id, client_secret = self.get_client_credentials()
        return {
            'id': self.id,
            'key': self.key,
            'name': self.name,
            'user_id': self.user_id,
            'created_by': self.created_by,
            'expiration_date': self.expiration_date.isoformat() if self.expiration_date else None,
            'created_at': self.created_at.isoformat() if self.created_at else None,
            'updated_at': self.updated_at.isoformat() if self.updated_at else None,
            'is_active': self.is_active,
            'is_expired': self.is_expired(),
            'client_id': client_id,
            'client_secret': client_secret
        }


class ExternalIdentity(db.Model):
    __tablename__ = 'external_identities'
    __table_args__ = (
        UniqueConstraint('provider', 'provider_id', 'external_id', name='uq_external_identity'),
        db.Index('ix_external_identities_user_id', 'user_id'),
        db.Index('ix_external_identities_tenant_id', 'tenant_id'),
    )

    id = db.Column(Integer, primary_key=True)
    provider = db.Column(String(64), nullable=False)  # e.g., 'saml'
    provider_id = db.Column(Integer, nullable=True)  # SamlConfiguration.id for SAML
    external_id = db.Column(String(255), nullable=False)
    tenant_id = db.Column(Integer, ForeignKey('tenants.id'), nullable=True)
    user_id = db.Column(Integer, ForeignKey('users.id'), nullable=False)
    attributes = db.Column(MutableDict.as_mutable(JSON), default=dict)
    last_login_at = db.Column(DateTime, nullable=True)
    created_at = db.Column(DateTime, default=datetime.utcnow)
    updated_at = db.Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    user = relationship('User', backref='external_identities')
    tenant = relationship('Tenant', backref='external_identities')

    def to_dict(self):
        return {
            'id': self.id,
            'provider': self.provider,
            'provider_id': self.provider_id,
            'external_id': self.external_id,
            'tenant_id': self.tenant_id,
            'user_id': self.user_id,
            'attributes': self.attributes,
            'last_login_at': self.last_login_at.isoformat() if self.last_login_at else None,
            'created_at': self.created_at.isoformat() if self.created_at else None,
            'updated_at': self.updated_at.isoformat() if self.updated_at else None,
        }


class SamlConfiguration(db.Model):
    __tablename__ = 'saml_configurations'
    __table_args__ = (
        db.Index('ix_saml_configurations_tenant_id', 'tenant_id'),
        db.Index('ix_saml_configurations_enabled', 'enabled'),
        UniqueConstraint('tenant_id', 'slug', name='uq_saml_tenant_slug'),
    )

    id = db.Column(Integer, primary_key=True)
    tenant_id = db.Column(Integer, ForeignKey('tenants.id'), nullable=False)
    name = db.Column(String(128), nullable=False)
    slug = db.Column(String(64), unique=True, nullable=False)
    enabled = db.Column(Boolean, default=True)
    metadata_xml = db.Column(Text, nullable=False)
    idp_entity_id = db.Column(String(512), nullable=False)
    idp_sso_url = db.Column(String(512), nullable=False)
    idp_slo_url = db.Column(String(512), nullable=True)
    idp_certificate = db.Column(Text, nullable=False)
    sp_entity_id = db.Column(String(512), nullable=True)
    acs_url = db.Column(String(512), nullable=True)
    slo_url = db.Column(String(512), nullable=True)
    attribute_mapping = db.Column(MutableDict.as_mutable(JSON), default=_default_attribute_mapping)
    auto_provision_users = db.Column(Boolean, default=True)
    allow_idp_initiated = db.Column(Boolean, default=True)
    created_at = db.Column(DateTime, default=datetime.utcnow)
    updated_at = db.Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    created_by = db.Column(Integer, ForeignKey('users.id'), nullable=True)

    tenant = relationship('Tenant', backref='saml_configurations')
    creator = relationship('User', foreign_keys=[created_by])

    def to_dict(self):
        return {
            'id': self.id,
            'tenant_id': self.tenant_id,
            'name': self.name,
            'slug': self.slug,
            'enabled': self.enabled,
            'idp_entity_id': self.idp_entity_id,
            'idp_sso_url': self.idp_sso_url,
            'idp_slo_url': self.idp_slo_url,
            'sp_entity_id': self.sp_entity_id,
            'acs_url': self.acs_url,
            'slo_url': self.slo_url,
            'attribute_mapping': self.attribute_mapping,
            'auto_provision_users': self.auto_provision_users,
            'allow_idp_initiated': self.allow_idp_initiated,
            'created_at': self.created_at.isoformat() if self.created_at else None,
            'updated_at': self.updated_at.isoformat() if self.updated_at else None,
        }

class Attestation(db.Model):
    __tablename__ = 'attestations'
    
    id = db.Column(Integer, primary_key=True, doc="Primary key")
    key = db.Column(String(255), nullable=False, doc="Unique identifier for the attestation")  # Unique identifier for the attestation
    target_name = db.Column(String(255), nullable=False, doc="Attestation Target name (e.g. image/file/repo name)")
    target_type = db.Column(String(100), nullable=False, doc="Target type (e.g. container, file, git, etc.)")
    content_type = db.Column(String(100), nullable=False, doc="Content type (e.g. cyclonedx-json)")
    context_type = db.Column(String(100), nullable=False, doc="Context type (e.g. local, github, gitlab, etc.)")
    tool = db.Column(String(100), doc="Tool that generated the attestation (usually valint)")
    sig_status = db.Column(String(50), default='unsigned', doc="Signature status (default 'unsigned')")
    deliverable = db.Column(Boolean, default=False, doc="Whether this attestation is of a deliverable artifact")
    
    # Context fields for deduplication
    product_key = db.Column(String(255), doc="Product key (name) for deduplication")
    product_version = db.Column(String(100), doc="Product version")
    labels = db.Column(JSON, doc="Labels as JSON array")  # Store as JSON array
    
    # Storage information
    storage_path = db.Column(String(1000), doc="File path or S3 key for storage")  # File path or S3 key
    storage_url = db.Column(String(3000), doc="Download URL for the attestation")  # Download URL
    file_size = db.Column(Integer, doc="File size in bytes")
    
    # Context data
    context = db.Column(JSON, doc="Complete context stored as JSON")  # Store complete context as JSON
    
    # Replacement tracking
    replaced_attestation_id = db.Column(Integer, ForeignKey('attestations.id'), nullable=True, doc="ID of the attestation that was replaced")
    replaced_by_attestation_id = db.Column(Integer, ForeignKey('attestations.id'), nullable=True, doc="ID of the attestation that replaced this one")
    
    # Metadata
    user_id = db.Column(Integer, ForeignKey('users.id'), doc="ID of the user who created the attestation")
    team_id = db.Column(Integer, doc="Team ID. Not in use yet.")
    job_ids = db.Column(JSON, doc="Job IDs as JSON array (backward compatibility)")  # Store as JSON array
    pipeline_run = db.Column(Integer, doc="Pipeline run number (backward compatibility)")
    upload_pending = db.Column(Boolean, default=False, doc="True if upload is pending")  # True if waiting for file upload
    
    # Timestamps
    created_at = db.Column(DateTime, default=datetime.utcnow, doc="Timestamp when the attestation was created")
    updated_at = db.Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, doc="Timestamp when the attestation was last updated")
    deleted = db.Column(Boolean, default=False, doc="True if the attestation is deleted")

    __table_args__ = (
        db.Index('ix_attestations_user_id_product_key', 'user_id', 'product_key'),
    )
    
    # Relationships
    user = relationship("User", backref=db.backref("attestations", cascade="all, delete-orphan"))
    
    def to_dict(self):
        return {
            'id': self.id,
            'key': self.key,
            'target_name': self.target_name,
            'target_type': self.target_type,
            'content_type': self.content_type,
            'context_type': self.context_type,
            'tool': self.tool,
            'sig_status': self.sig_status,
            'deliverable': self.deliverable,
            'product_key': self.product_key,
            'product_version': self.product_version,
            'labels': self.labels,
            'storage_url': self.storage_url,
            'file_size': self.file_size,
            'context': self.context,
            'user_id': self.user_id,
            'team_id': self.team_id,
            'job_ids': self.job_ids,
            'pipeline_run': self.pipeline_run,
            'replaced_attestation_id': self.replaced_attestation_id,
            'replaced_by_attestation_id': self.replaced_by_attestation_id,
            'created_at': self.created_at.isoformat() if self.created_at else None,
            'updated_at': self.updated_at.isoformat() if self.updated_at else None,
            'deleted': self.deleted
        }

class SystemLog(db.Model):
    __tablename__ = 'system_logs'
    __table_args__ = (
        db.Index('ix_system_logs_timestamp', 'timestamp'),
        db.Index('ix_system_logs_user_id', 'user_id'),
        db.Index('ix_system_logs_severity', 'severity'),
        db.Index('ix_system_logs_event_type', 'event_type'),
        db.Index('ix_system_logs_app_name', 'app_name'),
        db.Index('ix_system_logs_hostname', 'hostname'),
    )
    
    id = db.Column(Integer, primary_key=True)
    facility = db.Column(Integer, nullable=False)  # RFC 5424 facility
    severity = db.Column(Integer, nullable=False, index=False)  # RFC 5424 severity
    version = db.Column(Integer, default=1)  # RFC 5424 version
    timestamp = db.Column(DateTime, default=datetime.utcnow, index=False)
    hostname = db.Column(String(255), index=False)
    app_name = db.Column(String(48), index=False)
    proc_id = db.Column(String(128))
    msg_id = db.Column(String(32))
    structured_data = db.Column(Text)  # RFC 5424 structured data
    message = db.Column(Text)
    
    # Additional fields for internal use
    user_id = db.Column(Integer, ForeignKey('users.id'), nullable=True, index=False)
    source_ip = db.Column(String(45))  # IPv4 or IPv6
    event_type = db.Column(String(50), index=False)  # login, upload, download, etc.
    
    # Relationships
    user = relationship("User", backref="logs")
    
    def to_dict(self):
        return {
            'id': self.id,
            'facility': self.facility,
            'severity': self.severity,
            'version': self.version,
            'timestamp': self.timestamp.isoformat() if self.timestamp else None,
            'hostname': self.hostname,
            'app_name': self.app_name,
            'proc_id': self.proc_id,
            'msg_id': self.msg_id,
            'structured_data': self.structured_data,
            'message': self.message,
            'user_id': self.user_id,
            'source_ip': self.source_ip,
            'event_type': self.event_type
        }
    
    def to_rfc5424(self):
        """Convert to RFC 5424 format"""
        priority = self.facility * 8 + self.severity
        timestamp_str = self.timestamp.strftime('%Y-%m-%dT%H:%M:%S.%fZ') if self.timestamp else '-'
        hostname = self.hostname or '-'
        app_name = self.app_name or '-'
        proc_id = self.proc_id or '-'
        msg_id = self.msg_id or '-'
        structured_data = self.structured_data or '-'
        message = self.message or ''
        
        return f"<{priority}>{self.version} {timestamp_str} {hostname} {app_name} {proc_id} {msg_id} {structured_data} {message}"

class UploadSession(db.Model):
    __tablename__ = 'upload_sessions'
    
    id = db.Column(Integer, primary_key=True)
    session_id = db.Column(String(255), unique=True, nullable=False)
    user_id = db.Column(Integer, ForeignKey('users.id'))
    status = db.Column(String(50), default='pending')  # pending, uploading, completed, failed
    file_name = db.Column(String(255))
    file_size = db.Column(Integer)
    upload_url = db.Column(String(3000))  # Presigned URL for upload
    storage_path = db.Column(String(1000))
    error_message = db.Column(Text)
    created_at = db.Column(DateTime, default=datetime.utcnow)
    completed_at = db.Column(DateTime)
    
    # Relationships
    user = relationship("User", backref="upload_sessions")
    
    def to_dict(self):
        return {
            'id': self.id,
            'session_id': self.session_id,
            'user_id': self.user_id,
            'status': self.status,
            'file_name': self.file_name,
            'file_size': self.file_size,
            'upload_url': self.upload_url,
            'storage_path': self.storage_path,
            'error_message': self.error_message,
            'created_at': self.created_at.isoformat() if self.created_at else None,
            'completed_at': self.completed_at.isoformat() if self.completed_at else None
        }

class Webhook(db.Model):
    __tablename__ = 'webhooks'

    id = db.Column(Integer, primary_key=True)
    name = db.Column(String(255), nullable=False, unique=True)
    # Trigger type: "badge" or "attestation-condition"
    trigger_type = db.Column(String(64), nullable=False, default="badge")
    # Badge event trigger: store badge name and whether to trigger on non-zero count
    badge = db.Column(String(255), nullable=True)
    trigger_non_zero = db.Column(Boolean, default=True)
    # Attestation-condition: list of {field, value} pairs (ANDed)
    attestation_conditions = db.Column(JSON, nullable=True)
    url = db.Column(String(2000), nullable=False)
    secret = db.Column(String(255), nullable=True)
    created_by = db.Column(Integer, ForeignKey('users.id'), nullable=False)
    created_at = db.Column(DateTime, default=datetime.utcnow)
    is_active = db.Column(Boolean, default=True)
    headers = db.Column(db.Text, nullable=True)  # JSON string
    content = db.Column(db.Text, nullable=True)  # JSON string/template
    method = db.Column(String(10), nullable=False, default='POST')

    user = relationship("User", backref="webhooks")

    def to_dict(self):
        return {
            'id': self.id,
            'name': self.name,
            'trigger_type': self.trigger_type,
            'badge': self.badge,
            'trigger_non_zero': self.trigger_non_zero,
            'attestation_conditions': self.attestation_conditions,
            'url': self.url,
            'secret': self.secret,
            'created_by': self.created_by,
            'created_at': self.created_at.isoformat() if self.created_at else None,
            'is_active': self.is_active,
            'headers': self.headers,
            'content': self.content,
            'method': self.method
        }
# --- Policy Result and RuleResult Models ---
class Product(db.Model):
    __tablename__ = 'products'
    id = db.Column(Integer, primary_key=True)
    product_key = db.Column(String(255), nullable=False)
    product_version = db.Column(String(255), nullable=False)
    user_id = db.Column(Integer, ForeignKey('users.id'), nullable=False)
    name = db.Column(String(255), nullable=True)
    description = db.Column(Text, nullable=True)
    product_metadata = db.Column(MutableDict.as_mutable(JSON), nullable=True)
    badges = db.Column(MutableDict.as_mutable(JSON), nullable=True)  # Store badge data as JSON
    created_at = db.Column(DateTime, default=datetime.utcnow)
    updated_at = db.Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    deleted = db.Column(Boolean, default=False)

    user = relationship("User", backref="products")

    __table_args__ = (
        db.UniqueConstraint('user_id', 'product_key', 'product_version', name='uq_user_product_version'),
        db.Index('ix_products_product_key', 'product_key'),
    )

    def to_dict(self):
        return {
            'id': self.id,
            'product_key': self.product_key,
            'product_version': self.product_version,
            'user_id': self.user_id,
            'name': self.name,
            'description': self.description,
            'product_metadata': self.product_metadata,
            'badges': self.badges,
            'created_at': self.created_at.isoformat() if self.created_at else None,
            'updated_at': self.updated_at.isoformat() if self.updated_at else None,
            'deleted': self.deleted
        }

class PolicyResult(db.Model):
    __tablename__ = 'policy_results'
    id = db.Column(Integer, primary_key=True, doc="Primary key")
    attestation_id = db.Column(Integer, ForeignKey('attestations.id'), nullable=False, doc="Related attestation ID")
    user_id = db.Column(Integer, ForeignKey('users.id'), nullable=True, doc="User ID who created the attestation")  # User who created the attestation
    attestation_timestamp = db.Column(DateTime, nullable=False, doc="Timestamp of the attestation")
    initiative_id = db.Column(String(255), nullable=False, doc="Policy initiative ID")
    rule_result_id = db.Column(String(255), nullable=False, doc="Policy rule result ID")
    rule_id = db.Column(String(255), nullable=False, doc="Policy rule ID")
    rule_name = db.Column(String(255), nullable=True, doc="Name of the policy rule")
    rule_full_description = db.Column(Text, nullable=True, doc="Full description of the policy rule")
    result_kind = db.Column(String(64), nullable=True, doc="Type of the result")
    result_level = db.Column(String(64), nullable=True, doc="Level of the result")
    result_message = db.Column(Text, nullable=True, doc="Message for the result")
    rule_json = db.Column(JSON, nullable=False, doc="JSON of rule metadata")
    result_json = db.Column(JSON, nullable=False, doc="JSON of result metadata")
    product_key = db.Column(String(255), nullable=True, doc="Product key")
    product_version = db.Column(String(255), nullable=True, doc="Product version")
    deleted = db.Column(Boolean, default=False, doc="True if the policy result is deleted")

    attestation = relationship("Attestation", backref=db.backref("policy_results", cascade="all, delete-orphan"))
    user = relationship("User", backref=db.backref("policy_results", cascade="all, delete-orphan"))

    __table_args__ = (
        db.Index('ix_policy_results_user_id_product_key', 'user_id', 'product_key'),
        db.Index('ix_policy_results_attestation_timestamp', 'attestation_timestamp'),
    )

    def to_dict(self):
        return {
            'id': self.id,
            'attestation_id': self.attestation_id,
            'user_id': self.user_id,
            'attestation_timestamp': self.attestation_timestamp.isoformat() if self.attestation_timestamp else None,
            'initiative_id': self.initiative_id,
            'rule_result_id': self.rule_result_id,
            'rule_id': self.rule_id,
            'rule_name': self.rule_name,
            'rule_full_description': self.rule_full_description,
            'result_kind': self.result_kind,
            'result_level': self.result_level,
            'result_message': self.result_message,
            'rule_json': self.rule_json,
            'result_json': self.result_json,
            'deleted': self.deleted
        }