mirror of
https://github.com/taylanbakircioglu/haproxy-openmanager.git
synced 2026-09-16 15:45:11 +00:00
taylanbakircioglu
9e2ea04777
Bulk import / manual edit silently dropped `filter spoe engine ...` (Coraza WAF) and frontend `log-format` because the parser recognised only a fixed directive set. The regenerated config then missed the SPOE engine, so HAProxy failed with "unable to find SPOE engine 'coraza' used by the send-spoe-group". - parser: capture `filter` + `log-format`/`log-format-sd` into new ParsedFrontend fields - db: additive nullable `log_format` + `filters` TEXT columns on frontends (SCHEMA_VERSION 8->9) - generator: new `filter` bucket flushed before http-request rules so `filter` precedes `send-spoe-group`; `log-format` kept in prelude - bulk import: preview dict, change-detection, persist (create + merge-update); cluster-aware SPOE pre-flight advisories (missing-filter + host-prerequisite) surfaced in the UI - manual CRUD: full round-trip (get/create/update) incl. React form fields (no null-wipe) - reject/rollback: restore the new columns; restore path + wizard helper kept in parity - backend `option spop-check` recognised (suppresses spurious warning for coraza-spoa) - tests: test_spoe_filter_import.py; full suite green (1079 passed)
685 lines
29 KiB
Python
685 lines
29 KiB
Python
"""
|
||
Entity Snapshot Module for HAProxy OpenManager
|
||
|
||
This module captures snapshots of entity changes so they can be
|
||
restored on reject of a config_version.
|
||
|
||
Usage:
|
||
# Capture an UPDATE snapshot.
|
||
old_entity = await conn.fetchrow("SELECT * FROM frontends WHERE id = 5")
|
||
snapshot_metadata = await save_entity_snapshot(
|
||
conn=conn,
|
||
entity_type="frontend",
|
||
entity_id=5,
|
||
old_values=old_entity,
|
||
new_values={"bind_port": 443},
|
||
operation="UPDATE"
|
||
)
|
||
|
||
# Persist the snapshot on the config version.
|
||
await conn.execute(
|
||
"INSERT INTO config_versions (..., metadata) VALUES (..., $1)",
|
||
json.dumps(snapshot_metadata)
|
||
)
|
||
|
||
# Roll back when the config version is rejected.
|
||
await rollback_entity_from_snapshot(conn, snapshot_metadata["entity_snapshot"])
|
||
|
||
Supported Operations:
|
||
- UPDATE: restore the entity's fields to their pre-change values
|
||
- CREATE: delete the newly-created entity (used by bulk import / wizard)
|
||
- UPDATE_RESTORE: undo an UPDATE performed during a restore flow
|
||
- DELETE: re-create a soft-deleted entity (currently unused — soft
|
||
delete is preferred)
|
||
|
||
Feature Flag:
|
||
ENTITY_SNAPSHOT_ENABLED=true|false
|
||
Default: false — safe-by-default until the operator opts in.
|
||
|
||
Author: Taylan Bakırcıoğlu
|
||
Date: 2025-01-13
|
||
Version: 1.0.0
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import logging
|
||
from typing import Dict, Any, Optional, List
|
||
from datetime import datetime
|
||
|
||
# Setup logger
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Entity snapshot enabled by default
|
||
# Can be disabled with ENTITY_SNAPSHOT_ENABLED=false if needed
|
||
ENTITY_SNAPSHOT_ENABLED = os.getenv("ENTITY_SNAPSHOT_ENABLED", "true").lower() == "true"
|
||
|
||
|
||
async def save_entity_snapshot(
|
||
conn,
|
||
entity_type: str,
|
||
entity_id: int,
|
||
old_values: Dict[str, Any],
|
||
new_values: Optional[Dict[str, Any]] = None,
|
||
operation: str = "UPDATE"
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
Entity snapshot'ı metadata formatında hazırla.
|
||
|
||
Args:
|
||
conn: Database connection (asyncpg)
|
||
entity_type: Entity tipi ('frontend', 'backend', 'waf_rule', 'ssl_certificate', 'server')
|
||
entity_id: Entity ID
|
||
old_values: Eski değerler (asyncpg Record'dan dict)
|
||
new_values: Yeni değerler (opsiyonel, UPDATE için)
|
||
operation: İşlem tipi ('CREATE', 'UPDATE', 'DELETE', 'UPDATE_RESTORE')
|
||
|
||
Returns:
|
||
Dict containing entity_snapshot metadata:
|
||
{
|
||
"entity_snapshot": {
|
||
"entity_type": "frontend",
|
||
"entity_id": 5,
|
||
"operation": "UPDATE",
|
||
"timestamp": "2025-01-13T10:30:45Z",
|
||
"old_values": {...},
|
||
"new_values": {...}, # Sadece değişen alanlar (compaction)
|
||
"changed_fields": ["bind_port", "ssl_enabled"]
|
||
}
|
||
}
|
||
|
||
Example:
|
||
old_frontend = await conn.fetchrow("SELECT * FROM frontends WHERE id = 5")
|
||
metadata = await save_entity_snapshot(
|
||
conn=conn,
|
||
entity_type="frontend",
|
||
entity_id=5,
|
||
old_values=dict(old_frontend),
|
||
new_values={"bind_port": 443, "ssl_enabled": True},
|
||
operation="UPDATE"
|
||
)
|
||
"""
|
||
# Feature flag check
|
||
if not ENTITY_SNAPSHOT_ENABLED:
|
||
logger.debug(f"Entity snapshot disabled by feature flag for {entity_type} {entity_id}")
|
||
return {}
|
||
|
||
try:
|
||
# Convert asyncpg Record to dict if needed
|
||
if hasattr(old_values, '__iter__') and not isinstance(old_values, dict):
|
||
old_values = dict(old_values)
|
||
|
||
# CRITICAL FIX: Convert non-JSON-serializable types to JSON-safe format
|
||
# Use try-catch for each value to handle asyncpg types, datetime, etc.
|
||
import json as json_test
|
||
serializable_old_values = {}
|
||
|
||
for key, value in old_values.items():
|
||
try:
|
||
# Test if value is JSON serializable
|
||
json_test.dumps(value)
|
||
# If no exception, use as-is
|
||
serializable_old_values[key] = value
|
||
except (TypeError, ValueError):
|
||
# Value is not JSON serializable, convert it
|
||
if value is None:
|
||
serializable_old_values[key] = None
|
||
elif hasattr(value, 'isoformat'):
|
||
# datetime, date, time objects
|
||
serializable_old_values[key] = str(value)
|
||
else:
|
||
# Everything else: convert to string
|
||
serializable_old_values[key] = str(value)
|
||
|
||
# CRITICAL FIX: Serialize new_values the same way as old_values
|
||
# Without this, datetime objects in new_values (e.g., expiry_date from SSL update)
|
||
# cause the final JSON serialization test to FAIL, resulting in empty snapshot ({})
|
||
# which means reject/rollback can NEVER find the snapshot to restore old values
|
||
serializable_new_values = {}
|
||
if new_values:
|
||
for key, value in new_values.items():
|
||
try:
|
||
json_test.dumps(value)
|
||
serializable_new_values[key] = value
|
||
except (TypeError, ValueError):
|
||
if value is None:
|
||
serializable_new_values[key] = None
|
||
elif hasattr(value, 'isoformat'):
|
||
serializable_new_values[key] = str(value)
|
||
else:
|
||
serializable_new_values[key] = str(value)
|
||
|
||
# Calculate changed fields for UPDATE operations
|
||
# Compare serialized old_values with serialized new_values for consistent comparison
|
||
changed_fields = []
|
||
if operation in ["UPDATE", "UPDATE_RESTORE"] and serializable_new_values:
|
||
changed_fields = [
|
||
field for field, new_val in serializable_new_values.items()
|
||
if field in serializable_old_values and serializable_old_values[field] != new_val
|
||
]
|
||
|
||
# Build snapshot
|
||
snapshot = {
|
||
"entity_type": entity_type,
|
||
"entity_id": entity_id,
|
||
"operation": operation,
|
||
"timestamp": datetime.utcnow().isoformat() + "Z",
|
||
"old_values": serializable_old_values, # JSON-safe values
|
||
"changed_fields": changed_fields
|
||
}
|
||
|
||
# For UPDATE: add new_values (compaction - only changed fields)
|
||
if operation in ["UPDATE", "UPDATE_RESTORE"] and serializable_new_values:
|
||
# Only store changed fields to save space (compaction)
|
||
snapshot["new_values"] = {
|
||
field: serializable_new_values[field]
|
||
for field in changed_fields
|
||
}
|
||
|
||
logger.info(
|
||
f"SNAPSHOT: Created for {entity_type} {entity_id} "
|
||
f"(operation={operation}, changed_fields={len(changed_fields)})"
|
||
)
|
||
logger.info(f"SNAPSHOT DEBUG: old_values keys={list(serializable_old_values.keys())[:10]}")
|
||
logger.info(f"SNAPSHOT DEBUG: Serializable check - can serialize to JSON: {len(serializable_old_values)} fields")
|
||
|
||
# Test if snapshot is JSON serializable
|
||
try:
|
||
import json as json_final_test
|
||
json_final_test.dumps(snapshot)
|
||
logger.info(f"SNAPSHOT DEBUG: Final JSON test PASSED for {entity_type} {entity_id}")
|
||
except Exception as json_err:
|
||
logger.error(f"SNAPSHOT DEBUG: Final JSON test FAILED for {entity_type} {entity_id}: {json_err}")
|
||
return {}
|
||
|
||
return {"entity_snapshot": snapshot}
|
||
|
||
except Exception as e:
|
||
logger.error(f"SNAPSHOT ERROR: Failed to create snapshot for {entity_type} {entity_id}: {e}")
|
||
# Return empty dict on error (graceful degradation)
|
||
return {}
|
||
|
||
|
||
async def rollback_entity_from_snapshot(
|
||
conn,
|
||
entity_snapshot: Dict[str, Any]
|
||
) -> bool:
|
||
"""
|
||
Entity'yi snapshot'taki eski değerlerine geri yükle.
|
||
|
||
Args:
|
||
conn: Database connection (asyncpg)
|
||
entity_snapshot: Snapshot dict (from metadata.entity_snapshot)
|
||
|
||
Returns:
|
||
True if rollback successful, False otherwise
|
||
|
||
Example:
|
||
metadata = json.loads(version['metadata'])
|
||
entity_snapshot = metadata.get('entity_snapshot')
|
||
|
||
if entity_snapshot:
|
||
success = await rollback_entity_from_snapshot(conn, entity_snapshot)
|
||
if success:
|
||
logger.info("Rollback successful")
|
||
else:
|
||
logger.warning("Rollback failed")
|
||
"""
|
||
if not ENTITY_SNAPSHOT_ENABLED:
|
||
logger.warning("ROLLBACK DEBUG: Entity snapshot disabled by feature flag, skipping rollback")
|
||
return False
|
||
|
||
entity_type = entity_snapshot.get("entity_type")
|
||
entity_id = entity_snapshot.get("entity_id")
|
||
operation = entity_snapshot.get("operation")
|
||
old_values = entity_snapshot.get("old_values")
|
||
|
||
logger.info(f"ROLLBACK DEBUG: entity_type={entity_type}, entity_id={entity_id}, operation={operation}")
|
||
logger.info(f"ROLLBACK DEBUG: old_values exists={old_values is not None}, old_values length={len(old_values) if old_values else 0}")
|
||
|
||
# R18 audit fix: the historical guard `not all([..., old_values])`
|
||
# treated `old_values={}` as missing because `{}` is falsy in Python.
|
||
# The wizard's bulk_snapshots always emit `"old_values": {}` for
|
||
# CREATE entries (there's nothing to restore on rollback — the
|
||
# rollback path is a DELETE) — so EVERY wizard CREATE snapshot
|
||
# silently bypassed the rollback shim. Rejecting a wizard PENDING
|
||
# version then visibly removed the config_versions row but left the
|
||
# wizard-created backends/servers/frontends/SSL certs orphaned in
|
||
# the DB. Fix: only require old_values for UPDATE/DELETE; for
|
||
# CREATE the field is intentionally empty.
|
||
if not all([entity_type, entity_id, operation]):
|
||
logger.warning(
|
||
"ROLLBACK: Invalid snapshot data, skipping rollback (missing: "
|
||
f"{[k for k in ['entity_type', 'entity_id', 'operation'] if not entity_snapshot.get(k)]})"
|
||
)
|
||
return False
|
||
if operation in ("UPDATE", "UPDATE_RESTORE", "DELETE") and not old_values:
|
||
logger.warning(
|
||
f"ROLLBACK: {operation} snapshot for {entity_type} {entity_id} "
|
||
"is missing old_values — cannot restore prior state"
|
||
)
|
||
return False
|
||
|
||
try:
|
||
if operation in ["UPDATE", "UPDATE_RESTORE"]:
|
||
# Entity'yi eski değerlerine geri yükle
|
||
logger.info(f"ROLLBACK DEBUG: Calling _rollback_update for {entity_type} {entity_id}")
|
||
success = await _rollback_update(conn, entity_type, entity_id, old_values)
|
||
logger.info(f"ROLLBACK DEBUG: _rollback_update returned {success}")
|
||
return success
|
||
|
||
elif operation == "CREATE":
|
||
# Yeni oluşturulan entity'yi sil
|
||
# ÖNEMLİ: Sadece bulk import ile TAMAMEN YENİ oluşturulan entity'ler için!
|
||
success = await _rollback_create(conn, entity_type, entity_id)
|
||
return success
|
||
|
||
elif operation == "DELETE":
|
||
# Silinen entity'yi geri yükle
|
||
# NOT: Şu an soft-delete kullanıldığı için kullanılmıyor
|
||
success = await _rollback_delete(conn, entity_type, entity_id, old_values)
|
||
return success
|
||
|
||
else:
|
||
logger.warning(f"ROLLBACK: Unknown operation '{operation}', skipping")
|
||
return False
|
||
|
||
except Exception as e:
|
||
logger.error(f"ROLLBACK ERROR: Failed for {entity_type} {entity_id}: {e}")
|
||
return False
|
||
|
||
|
||
async def _rollback_update(
|
||
conn,
|
||
entity_type: str,
|
||
entity_id: int,
|
||
old_values: Dict[str, Any]
|
||
) -> bool:
|
||
"""
|
||
Entity'yi UPDATE öncesi haline döndür.
|
||
|
||
ÖNEMLİ: Bu fonksiyon var olan entity'lerin field'larını eski değerlere döndürür.
|
||
Entity'yi KESİNLİKLE silmez!
|
||
|
||
Args:
|
||
conn: Database connection
|
||
entity_type: Entity tipi
|
||
entity_id: Entity ID
|
||
old_values: Eski değerler (snapshot'tan)
|
||
|
||
Returns:
|
||
True if successful, False otherwise
|
||
"""
|
||
try:
|
||
if entity_type == "frontend":
|
||
# Frontend'i eski değerlerine geri yükle
|
||
logger.info(f"ROLLBACK UPDATE DEBUG: Frontend {entity_id} - restoring bind_port={old_values.get('bind_port')}")
|
||
logger.info(f"ROLLBACK UPDATE DEBUG: old_values sample: name={old_values.get('name')}, bind_port={old_values.get('bind_port')}, ssl_enabled={old_values.get('ssl_enabled')}")
|
||
|
||
result = await conn.execute("""
|
||
UPDATE frontends SET
|
||
name = $1, bind_address = $2, bind_port = $3,
|
||
default_backend = $4, mode = $5, ssl_enabled = $6,
|
||
ssl_certificate_id = $7, ssl_certificate_ids = $8, ssl_port = $9,
|
||
ssl_cert_path = $10, ssl_cert = $11, ssl_verify = $12,
|
||
ssl_alpn = $13, ssl_npn = $14, ssl_ciphers = $15, ssl_ciphersuites = $16,
|
||
ssl_min_ver = $17, ssl_max_ver = $18, ssl_strict_sni = $19,
|
||
acl_rules = $20, redirect_rules = $21, use_backend_rules = $22,
|
||
request_headers = $23, response_headers = $24, options = $25,
|
||
tcp_request_rules = $26, timeout_client = $27, timeout_http_request = $28,
|
||
rate_limit = $29, compression = $30, log_separate = $31,
|
||
monitor_uri = $32, maxconn = $33,
|
||
cluster_id = $34, is_active = $35, last_config_status = $36,
|
||
log_format = $37, filters = $38,
|
||
updated_at = CURRENT_TIMESTAMP
|
||
WHERE id = $39
|
||
""",
|
||
old_values.get('name'),
|
||
old_values.get('bind_address'),
|
||
old_values.get('bind_port'),
|
||
old_values.get('default_backend'),
|
||
old_values.get('mode'),
|
||
old_values.get('ssl_enabled'),
|
||
old_values.get('ssl_certificate_id'),
|
||
old_values.get('ssl_certificate_ids'),
|
||
old_values.get('ssl_port'),
|
||
old_values.get('ssl_cert_path'),
|
||
old_values.get('ssl_cert'),
|
||
old_values.get('ssl_verify'),
|
||
old_values.get('ssl_alpn'),
|
||
old_values.get('ssl_npn'),
|
||
old_values.get('ssl_ciphers'),
|
||
old_values.get('ssl_ciphersuites'),
|
||
old_values.get('ssl_min_ver'),
|
||
old_values.get('ssl_max_ver'),
|
||
old_values.get('ssl_strict_sni'),
|
||
old_values.get('acl_rules'),
|
||
old_values.get('redirect_rules'),
|
||
old_values.get('use_backend_rules'),
|
||
old_values.get('request_headers'),
|
||
old_values.get('response_headers'),
|
||
old_values.get('options'),
|
||
old_values.get('tcp_request_rules'),
|
||
old_values.get('timeout_client'),
|
||
old_values.get('timeout_http_request'),
|
||
old_values.get('rate_limit'),
|
||
old_values.get('compression'),
|
||
old_values.get('log_separate'),
|
||
old_values.get('monitor_uri'),
|
||
old_values.get('maxconn'),
|
||
old_values.get('cluster_id'),
|
||
old_values.get('is_active'),
|
||
old_values.get('last_config_status'),
|
||
old_values.get('log_format'), # Issue #38
|
||
old_values.get('filters'), # Issue #38
|
||
entity_id
|
||
)
|
||
|
||
logger.info(f"ROLLBACK UPDATE: Frontend {entity_id} restored to previous state (UPDATE query result={result})")
|
||
|
||
# Verify rollback
|
||
verify = await conn.fetchrow("SELECT bind_port, last_config_status FROM frontends WHERE id = $1", entity_id)
|
||
logger.info(f"ROLLBACK UPDATE VERIFY: Frontend {entity_id} after rollback - bind_port={verify['bind_port']}, status={verify['last_config_status']}")
|
||
|
||
return True
|
||
|
||
elif entity_type == "backend":
|
||
# Backend'i eski değerlerine geri yükle
|
||
# SCHEMA: backends (ALL FIELDS from migrations.py line 1943-1962 + additions 1966-2006)
|
||
await conn.execute("""
|
||
UPDATE backends SET
|
||
name = $1, balance_method = $2, mode = $3,
|
||
health_check_uri = $4, health_check_interval = $5,
|
||
health_check_expected_status = $6, fullconn = $7,
|
||
timeout_connect = $8, timeout_server = $9, timeout_queue = $10,
|
||
cluster_id = $11, maxconn = $12,
|
||
cookie_name = $13, cookie_options = $14,
|
||
default_server_inter = $15, default_server_fall = $16,
|
||
default_server_rise = $17, request_headers = $18,
|
||
response_headers = $19, options = $20,
|
||
is_active = $21, last_config_status = $22,
|
||
updated_at = CURRENT_TIMESTAMP
|
||
WHERE id = $23
|
||
""",
|
||
old_values.get('name'),
|
||
old_values.get('balance_method'),
|
||
old_values.get('mode'),
|
||
old_values.get('health_check_uri'),
|
||
old_values.get('health_check_interval'),
|
||
old_values.get('health_check_expected_status'),
|
||
old_values.get('fullconn'),
|
||
old_values.get('timeout_connect'),
|
||
old_values.get('timeout_server'),
|
||
old_values.get('timeout_queue'),
|
||
old_values.get('cluster_id'),
|
||
old_values.get('maxconn'),
|
||
old_values.get('cookie_name'),
|
||
old_values.get('cookie_options'),
|
||
old_values.get('default_server_inter'),
|
||
old_values.get('default_server_fall'),
|
||
old_values.get('default_server_rise'),
|
||
old_values.get('request_headers'),
|
||
old_values.get('response_headers'),
|
||
old_values.get('options'),
|
||
old_values.get('is_active'),
|
||
old_values.get('last_config_status'),
|
||
entity_id
|
||
)
|
||
logger.info(f"ROLLBACK UPDATE: Backend {entity_id} restored to previous state")
|
||
return True
|
||
|
||
elif entity_type == "waf_rule":
|
||
# WAF rule'u eski değerlerine geri yükle
|
||
# SCHEMA: waf_rules (ALL FIELDS from migrations.py line 2144-2160)
|
||
await conn.execute("""
|
||
UPDATE waf_rules SET
|
||
name = $1, rule_type = $2, config = $3, action = $4,
|
||
priority = $5, description = $6, enabled = $7,
|
||
cluster_id = $8, is_active = $9, last_config_status = $10,
|
||
updated_at = CURRENT_TIMESTAMP
|
||
WHERE id = $11
|
||
""",
|
||
old_values.get('name'),
|
||
old_values.get('rule_type'),
|
||
old_values.get('config'),
|
||
old_values.get('action'),
|
||
old_values.get('priority'),
|
||
old_values.get('description'),
|
||
old_values.get('enabled'),
|
||
old_values.get('cluster_id'),
|
||
old_values.get('is_active'),
|
||
old_values.get('last_config_status'),
|
||
entity_id
|
||
)
|
||
logger.info(f"ROLLBACK UPDATE: WAF rule {entity_id} restored to previous state")
|
||
return True
|
||
|
||
elif entity_type == "ssl_certificate":
|
||
# SSL certificate'i eski degerlerine geri yukle
|
||
# expiry_date snapshot'ta string olarak saklanir, datetime'a cevirilmesi gerekir
|
||
expiry_date_val = None
|
||
raw_expiry = old_values.get('expiry_date')
|
||
if raw_expiry:
|
||
try:
|
||
from datetime import datetime as dt_parse
|
||
if isinstance(raw_expiry, str):
|
||
# Parse ISO format string back to datetime
|
||
clean = raw_expiry.replace('Z', '+00:00')
|
||
expiry_date_val = dt_parse.fromisoformat(clean).replace(tzinfo=None)
|
||
else:
|
||
expiry_date_val = raw_expiry
|
||
except Exception as ed_err:
|
||
logger.warning(f"ROLLBACK: Could not parse expiry_date '{raw_expiry}': {ed_err}, skipping expiry_date restore")
|
||
expiry_date_val = None
|
||
|
||
await conn.execute("""
|
||
UPDATE ssl_certificates SET
|
||
name = $1, primary_domain = $2, certificate_content = $3,
|
||
private_key_content = $4, chain_content = $5,
|
||
issuer = $6, status = $7,
|
||
fingerprint = $8, days_until_expiry = $9, all_domains = $10,
|
||
cluster_id = $11, last_config_status = $12, usage_type = $13,
|
||
is_active = $14, expiry_date = $15,
|
||
updated_at = CURRENT_TIMESTAMP
|
||
WHERE id = $16
|
||
""",
|
||
old_values.get('name'),
|
||
old_values.get('primary_domain'),
|
||
old_values.get('certificate_content'),
|
||
old_values.get('private_key_content'),
|
||
old_values.get('chain_content'),
|
||
old_values.get('issuer'),
|
||
old_values.get('status'),
|
||
old_values.get('fingerprint'),
|
||
old_values.get('days_until_expiry'),
|
||
old_values.get('all_domains'),
|
||
old_values.get('cluster_id'),
|
||
old_values.get('last_config_status'),
|
||
old_values.get('usage_type'),
|
||
old_values.get('is_active'),
|
||
expiry_date_val,
|
||
entity_id
|
||
)
|
||
logger.info(f"ROLLBACK UPDATE: SSL certificate {entity_id} restored to previous state (including content and expiry)")
|
||
return True
|
||
|
||
elif entity_type == "cluster":
|
||
await conn.execute("""
|
||
UPDATE haproxy_clusters SET
|
||
acme_enabled = $1,
|
||
acme_backend_url = $2,
|
||
updated_at = CURRENT_TIMESTAMP
|
||
WHERE id = $3
|
||
""",
|
||
old_values.get('acme_enabled'),
|
||
old_values.get('acme_backend_url'),
|
||
entity_id
|
||
)
|
||
logger.info(f"ROLLBACK UPDATE: Cluster {entity_id} acme_enabled restored to {old_values.get('acme_enabled')}")
|
||
return True
|
||
|
||
elif entity_type == "server":
|
||
# Server'ı eski değerlerine geri yükle
|
||
# SCHEMA: backend_servers (ALL FIELDS from migrations.py line 2010-2036)
|
||
await conn.execute("""
|
||
UPDATE backend_servers SET
|
||
backend_id = $1, backend_name = $2, server_name = $3,
|
||
server_address = $4, server_port = $5, weight = $6,
|
||
maxconn = $7, check_enabled = $8, check_port = $9,
|
||
backup_server = $10, ssl_enabled = $11, ssl_verify = $12,
|
||
ssl_certificate_id = $13,
|
||
ssl_sni = $14, ssl_min_ver = $15, ssl_max_ver = $16, ssl_ciphers = $17,
|
||
cookie_value = $18,
|
||
inter = $19, fall = $20, rise = $21,
|
||
cluster_id = $22, is_active = $23, last_config_status = $24,
|
||
haproxy_status = $25,
|
||
updated_at = CURRENT_TIMESTAMP
|
||
WHERE id = $26
|
||
""",
|
||
old_values.get('backend_id'),
|
||
old_values.get('backend_name'),
|
||
old_values.get('server_name'),
|
||
old_values.get('server_address'),
|
||
old_values.get('server_port'),
|
||
old_values.get('weight'),
|
||
old_values.get('maxconn'),
|
||
old_values.get('check_enabled'),
|
||
old_values.get('check_port'),
|
||
old_values.get('backup_server'),
|
||
old_values.get('ssl_enabled'),
|
||
old_values.get('ssl_verify'),
|
||
old_values.get('ssl_certificate_id'),
|
||
old_values.get('ssl_sni'),
|
||
old_values.get('ssl_min_ver'),
|
||
old_values.get('ssl_max_ver'),
|
||
old_values.get('ssl_ciphers'),
|
||
old_values.get('cookie_value'),
|
||
old_values.get('inter'),
|
||
old_values.get('fall'),
|
||
old_values.get('rise'),
|
||
old_values.get('cluster_id'),
|
||
old_values.get('is_active'),
|
||
old_values.get('last_config_status'),
|
||
old_values.get('haproxy_status'),
|
||
entity_id
|
||
)
|
||
logger.info(f"ROLLBACK UPDATE: Server {entity_id} restored to previous state")
|
||
return True
|
||
|
||
else:
|
||
logger.warning(f"ROLLBACK UPDATE: Unsupported entity type '{entity_type}'")
|
||
return False
|
||
|
||
except Exception as e:
|
||
logger.error(f"ROLLBACK UPDATE ERROR: {entity_type} {entity_id}: {e}", exc_info=True)
|
||
return False
|
||
|
||
|
||
async def _rollback_create(
|
||
conn,
|
||
entity_type: str,
|
||
entity_id: int
|
||
) -> bool:
|
||
"""
|
||
Yeni oluşturulan entity'yi sil (reject edilen CREATE).
|
||
|
||
ÖNEMLİ: Bu fonksiyon SADECE bulk import ile TAMAMEN YENİ oluşturulan
|
||
entity'ler için kullanılır. Var olan entity'nin UPDATE'i için
|
||
KESİNLİKLE kullanılmaz!
|
||
|
||
Kullanım Senaryosu:
|
||
- Bulk import ile 5 yeni backend oluşturuldu
|
||
- Kullanıcı reject yaptı
|
||
- Bu fonksiyon 5 backend'i siler
|
||
|
||
Args:
|
||
conn: Database connection
|
||
entity_type: Entity tipi
|
||
entity_id: Entity ID (silinecek)
|
||
|
||
Returns:
|
||
True if successful, False otherwise
|
||
"""
|
||
try:
|
||
if entity_type == "frontend":
|
||
await conn.execute("DELETE FROM frontends WHERE id = $1", entity_id)
|
||
logger.info(f"ROLLBACK CREATE: Deleted frontend {entity_id}")
|
||
return True
|
||
|
||
elif entity_type == "backend":
|
||
# Cascade delete: servers otomatik silinecek (foreign key)
|
||
await conn.execute("DELETE FROM backends WHERE id = $1", entity_id)
|
||
logger.info(f"ROLLBACK CREATE: Deleted backend {entity_id} (+ cascade servers)")
|
||
return True
|
||
|
||
elif entity_type == "waf_rule":
|
||
await conn.execute("DELETE FROM waf_rules WHERE id = $1", entity_id)
|
||
logger.info(f"ROLLBACK CREATE: Deleted WAF rule {entity_id}")
|
||
return True
|
||
|
||
elif entity_type == "ssl_certificate":
|
||
await conn.execute("DELETE FROM ssl_certificates WHERE id = $1", entity_id)
|
||
logger.info(f"ROLLBACK CREATE: Deleted SSL certificate {entity_id}")
|
||
return True
|
||
|
||
elif entity_type == "server":
|
||
await conn.execute("DELETE FROM backend_servers WHERE id = $1", entity_id)
|
||
logger.info(f"ROLLBACK CREATE: Deleted server {entity_id}")
|
||
return True
|
||
|
||
elif entity_type == "letsencrypt_order":
|
||
# v1.5.0 Feature B: wizard-staged ACME orders attached to a
|
||
# bulk-site-create-* version (legacy: bulk-proxied-host-create-*).
|
||
# Cascade also removes acme_challenges (ON DELETE CASCADE).
|
||
await conn.execute(
|
||
"DELETE FROM letsencrypt_orders WHERE id = $1", entity_id
|
||
)
|
||
logger.info(
|
||
f"ROLLBACK CREATE: Deleted letsencrypt_order {entity_id} "
|
||
"(+ cascade acme_challenges)"
|
||
)
|
||
return True
|
||
|
||
else:
|
||
logger.warning(f"ROLLBACK CREATE: Unsupported entity type '{entity_type}'")
|
||
return False
|
||
|
||
except Exception as e:
|
||
logger.error(f"ROLLBACK CREATE ERROR: {entity_type} {entity_id}: {e}", exc_info=True)
|
||
return False
|
||
|
||
|
||
async def _rollback_delete(
|
||
conn,
|
||
entity_type: str,
|
||
entity_id: int,
|
||
old_values: Dict[str, Any]
|
||
) -> bool:
|
||
"""
|
||
Silinen entity'yi geri yükle (reject edilen DELETE).
|
||
|
||
NOT: Şu an projede soft-delete kullanıldığı için bu fonksiyon
|
||
aktif olarak kullanılmıyor. Gelecekte hard-delete kullanılırsa
|
||
bu fonksiyon implement edilecek.
|
||
|
||
Future Implementation:
|
||
- Entity'yi INSERT ile geri yükle
|
||
- Foreign key'leri geri yükle
|
||
- Related entity'leri geri yükle
|
||
|
||
Args:
|
||
conn: Database connection
|
||
entity_type: Entity tipi
|
||
entity_id: Entity ID
|
||
old_values: Eski değerler (snapshot'tan)
|
||
|
||
Returns:
|
||
True if successful, False otherwise
|
||
"""
|
||
logger.warning(
|
||
f"ROLLBACK DELETE: Not implemented yet for {entity_type} {entity_id}. "
|
||
"Currently using soft-delete (is_active=false). Hard-delete rollback is a future feature."
|
||
)
|
||
return False
|
||
|