"""Database Models""" from datetime import datetime from database import db from werkzeug.security import generate_password_hash, check_password_hash import json class User(db.Model): """User model for authentication""" __tablename__ = 'users' id = db.Column(db.Integer, primary_key=True) username = db.Column(db.String(80), unique=True, nullable=False, index=True) password_hash = db.Column(db.String(255), nullable=False) is_admin = db.Column(db.Boolean, default=False) created_at = db.Column(db.DateTime, default=datetime.utcnow) last_login = db.Column(db.DateTime) def set_password(self, password): """Hash and set password""" self.password_hash = generate_password_hash(password, method='pbkdf2:sha256') def check_password(self, password): """Verify password""" return check_password_hash(self.password_hash, password) def __repr__(self): return f'' class Certificate(db.Model): """SSL Certificate storage""" __tablename__ = 'certificates' id = db.Column(db.Integer, primary_key=True) name = db.Column(db.String(200), nullable=False, unique=True, index=True) cert_content = db.Column(db.Text, nullable=False) # Full PEM (cert + key combined) cert_only = db.Column(db.Text) # Separate cert (for info) key_only = db.Column(db.Text) # Separate key (for backup) # Metadata common_name = db.Column(db.String(255)) subject_alt_names = db.Column(db.Text) # JSON array issued_at = db.Column(db.DateTime) expires_at = db.Column(db.DateTime) created_at = db.Column(db.DateTime, default=datetime.utcnow, index=True) updated_at = db.Column(db.DateTime, onupdate=datetime.utcnow) # Relationships vhosts = db.relationship('VirtualHost', backref='certificate', lazy=True) def get_san_list(self): """Get Subject Alternative Names as list""" if self.subject_alt_names: try: return json.loads(self.subject_alt_names) except: return [] return [] def set_san_list(self, san_list): """Set Subject Alternative Names from list""" self.subject_alt_names = json.dumps(san_list) def __repr__(self): return f'' class VirtualHost(db.Model): """Virtual Host / Proxy configuration""" __tablename__ = 'virtual_hosts' id = db.Column(db.Integer, primary_key=True) name = db.Column(db.String(200), nullable=False, unique=True, index=True) hostname = db.Column(db.String(255), nullable=False) description = db.Column(db.Text) # ===== FRONTEND SETTINGS ===== frontend_ip = db.Column(db.String(50), default='0.0.0.0') frontend_port = db.Column(db.Integer, default=443) protocol = db.Column(db.String(10), default='http') # http or tcp # ===== SSL SETTINGS ===== use_ssl = db.Column(db.Boolean, default=False) certificate_id = db.Column(db.Integer, db.ForeignKey('certificates.id')) ssl_redirect = db.Column(db.Boolean, default=False) ssl_redirect_port = db.Column(db.Integer, default=80) # ===== LOAD BALANCING ===== lb_method = db.Column(db.String(50), default='roundrobin') # roundrobin, leastconn, source, uri # ===== SECURITY OPTIONS ===== dos_protection = db.Column(db.Boolean, default=False) dos_ban_duration = db.Column(db.String(20), default='30m') dos_limit_requests = db.Column(db.Integer, default=100) sql_injection_check = db.Column(db.Boolean, default=False) xss_check = db.Column(db.Boolean, default=False) webshell_check = db.Column(db.Boolean, default=False) # ===== HEADERS ===== add_custom_header = db.Column(db.Boolean, default=False) custom_header_name = db.Column(db.String(200)) custom_header_value = db.Column(db.String(500)) del_server_header = db.Column(db.Boolean, default=False) forward_for = db.Column(db.Boolean, default=True) # ===== STATE ===== enabled = db.Column(db.Boolean, default=True, index=True) # ===== TIMESTAMPS ===== created_at = db.Column(db.DateTime, default=datetime.utcnow, index=True) updated_at = db.Column(db.DateTime, onupdate=datetime.utcnow) # Relationships backend_servers = db.relationship('BackendServer', backref='vhost', lazy=True, cascade='all, delete-orphan') def __repr__(self): return f'' class BackendServer(db.Model): """Backend server for virtual host""" __tablename__ = 'backend_servers' id = db.Column(db.Integer, primary_key=True) vhost_id = db.Column(db.Integer, db.ForeignKey('virtual_hosts.id'), nullable=False) # Server info name = db.Column(db.String(100), nullable=False) ip_address = db.Column(db.String(50), nullable=False) port = db.Column(db.Integer, nullable=False) maxconn = db.Column(db.Integer) weight = db.Column(db.Integer, default=1) # Health check health_check = db.Column(db.Boolean, default=False) health_check_path = db.Column(db.String(200), default='/') # State enabled = db.Column(db.Boolean, default=True) created_at = db.Column(db.DateTime, default=datetime.utcnow) updated_at = db.Column(db.DateTime, onupdate=datetime.utcnow) def __repr__(self): return f'' class ConfigHistory(db.Model): """History of HAProxy configuration changes""" __tablename__ = 'config_history' id = db.Column(db.Integer, primary_key=True) config_content = db.Column(db.Text, nullable=False) change_type = db.Column(db.String(50)) # vhost_create, vhost_edit, vhost_delete, manual_edit vhost_id = db.Column(db.Integer, db.ForeignKey('virtual_hosts.id')) user_id = db.Column(db.Integer, db.ForeignKey('users.id')) description = db.Column(db.Text) created_at = db.Column(db.DateTime, default=datetime.utcnow, index=True) # Relationships vhost = db.relationship('VirtualHost') user = db.relationship('User') def __repr__(self): return f''