ameniboukattaya/CloudOps
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Change : sonarqubefix , backup.py

Browse Source
This commit is contained in:
ameniboukattaya
2026-06-21 19:31:24 +01:00
parent 32feb9d4eb
commit 04bbbb192b
1 changed files with 206 additions and 162 deletions
Download Patch File Download Diff File Expand all files Collapse all files

268
platform/modules/backups.py
View File

@@ -99,16 +99,18 @@ def get_vm_backups():
# ────────────────────────────────────────────────────────────────
# BACKUP HEALTH AUDIT
# ────────────────────────────────────────────────────────────────
#
# audit_backup() used to be one large function (cognitive complexity 18).
# It's now an orchestrator that calls one small, single-purpose helper per
# check. Each helper takes the shared `add()` callback and any state it
# needs, and is responsible for exactly one check — easy to read, easy to
# test, easy to extend without pushing complexity back up.
def audit_backup(backup_file, source='local'):
checks = []
def add(name, status, detail='', more=None):
entry = {'name': name, 'status': status, 'detail': detail}
if more:
entry['more'] = more
checks.append(entry)
def _resolve_archive_path(backup_file, source, add):
"""Figure out the local path to the archive, pulling it from the main
server first if needed. Returns the path, or None if it could not be
resolved (in which case an error result dict has already been added
via `add` and the caller should bail out)."""
if source == 'local':
archive_path = f"/root/backups/{backup_file}"
else:
@@ -123,34 +125,16 @@ def audit_backup(backup_file, source='local'):
f"{MAIN_SERVER_USER}@{MAIN_SERVER_IP}:/root/backups/{backup_file} "
f"{tmp_path}"
)
out, err = _run(pull_cmd, timeout=120)
_out, err = _run(pull_cmd, timeout=120)
if not os.path.exists(tmp_path):
return {
'ok': False, 'score': 0,
'backup_file': backup_file,
'file_size_bytes': None,
'file_size_display': None,
'health_tier': 'critical',
'health_label': 'Unhealthy',
'checks': [{'name': 'File Access', 'status': 'fail',
'detail': f'Could not pull from main server: {err}'}],
'summary': 'Cannot access backup file from this host.'
}
add('File Access', 'fail', f'Could not pull from main server: {err}')
return None
archive_path = tmp_path
if not os.path.exists(archive_path):
add('File Exists', 'fail', f'Not found: {archive_path}')
return {
'ok': False, 'score': 0, 'checks': checks,
'backup_file': backup_file,
'file_size_bytes': None,
'file_size_display': None,
'health_tier': 'critical',
'health_label': 'Unhealthy',
'summary': 'Backup file does not exist on disk.',
}
add('File Exists', 'pass', archive_path)
return archive_path
def _check_file_size(archive_path, add):
size_bytes = os.path.getsize(archive_path)
size_mb = size_bytes / (1024 * 1024)
size_human = _human_bytes(size_bytes)
@@ -159,17 +143,19 @@ def audit_backup(backup_file, source='local'):
'We flag archives under 1 MB as corrupt and under ~50 MB as unusually small for a full stack backup.',
]
if size_bytes < 1024 * 1024:
add('File Size', 'fail',
f'{size_human} — suspiciously tiny, likely corrupt', more=size_more)
add('File Size', 'fail', f'{size_human} — suspiciously tiny, likely corrupt', more=size_more)
elif size_mb < 50:
add('File Size', 'warn',
f'{size_human} — smaller than expected (typical full backup > 50 MB)', more=size_more)
add('File Size', 'warn', f'{size_human} — smaller than expected (typical full backup > 50 MB)', more=size_more)
else:
add('File Size', 'pass',
f'{size_human} — within expected range', more=size_more)
add('File Size', 'pass', f'{size_human} — within expected range', more=size_more)
return size_bytes, size_human
def _check_checksum(archive_path, add):
sha_file = archive_path + '.sha256'
if os.path.exists(sha_file):
if not os.path.exists(sha_file):
add('Checksum (SHA256)', 'warn', 'No .sha256 sidecar found — run a new backup to get checksums')
return
try:
with open(sha_file, 'r') as f:
expected_hash = f.read().split()[0].strip()
@@ -181,10 +167,9 @@ def audit_backup(backup_file, source='local'):
f'MISMATCH — expected {expected_hash[:20]}… got {actual_hash[:20]}…')
except Exception as e:
add('Checksum (SHA256)', 'warn', f'Could not verify: {e}')
else:
add('Checksum (SHA256)', 'warn',
'No .sha256 sidecar found — run a new backup to get checksums')
def _check_archive_integrity(archive_path, add):
try:
result = subprocess.run(
['gzip', '--test', archive_path],
@@ -198,6 +183,14 @@ def audit_backup(backup_file, source='local'):
add('Archive Integrity', 'fail',
f'gzip test failed: {(result.stderr or result.stdout)[:200]}')
except FileNotFoundError:
_check_archive_integrity_fallback(archive_path, add)
except subprocess.TimeoutExpired:
add('Archive Integrity', 'warn', 'Integrity check timed out — file is large, probably OK')
except Exception as e:
add('Archive Integrity', 'warn', f'Could not test: {e}')
def _check_archive_integrity_fallback(archive_path, add):
try:
import gzip
with gzip.open(archive_path, 'rb') as f:
@@ -205,39 +198,44 @@ def audit_backup(backup_file, source='local'):
add('Archive Integrity', 'pass', 'gzip header valid')
except Exception as e:
add('Archive Integrity', 'fail', f'Archive appears corrupt: {e}')
except subprocess.TimeoutExpired:
add('Archive Integrity', 'warn', 'Integrity check timed out — file is large, probably OK')
except Exception as e:
add('Archive Integrity', 'warn', f'Could not test: {e}')
members = []
def _list_archive_members(archive_path):
try:
with tarfile.open(archive_path, 'r:gz') as tf:
members = tf.getnames()
return tf.getnames()
except Exception:
pass
return []
def _check_internal_structure(members, add):
if not members:
add('Internal Structure', 'warn', 'Could not inspect archive members')
return
if members:
has_volumes = any('volumes/' in m for m in members)
has_info = any('backup-info.txt' in m for m in members)
has_compose = any('compose-files/' in m for m in members)
vol_count = len([m for m in members if '/volumes/' in m and m.endswith('.tar.gz')])
issues = []
if not has_volumes: issues.append('volumes/ missing')
if not has_info: issues.append('backup-info.txt missing')
if not has_volumes:
issues.append('volumes/ missing')
if not has_info:
issues.append('backup-info.txt missing')
if not issues:
detail = f'volumes/ ✓ backup-info.txt ✓'
if has_compose: detail += ' compose-files/ ✓'
if issues:
add('Internal Structure', 'fail', ' · '.join(issues))
return
detail = 'volumes/ ✓ backup-info.txt ✓'
if has_compose:
detail += ' compose-files/ ✓'
detail += f' ({vol_count} volume archives)'
add('Internal Structure', 'pass', detail)
else:
add('Internal Structure', 'fail', ' · '.join(issues))
else:
add('Internal Structure', 'warn', 'Could not inspect archive members')
SUSPICIOUS = [
_SUSPICIOUS_PATTERNS = [
(r'\.\./', 'path traversal (..)'),
(r'^/', 'absolute path in archive'),
(r'/etc/passwd', '/etc/passwd reference'),
@@ -246,43 +244,46 @@ def audit_backup(backup_file, source='local'):
(r'id_rsa(?!\.pub)', 'private SSH key reference'),
(r'authorized_keys', 'authorized_keys reference'),
]
def _check_security_scan(members, add):
found_suspicious = []
for m in members:
for pat, label in SUSPICIOUS:
for pat, label in _SUSPICIOUS_PATTERNS:
if re.search(pat, m):
found_suspicious.append(f'{m} ({label})')
break
if found_suspicious:
add('Security Scan', 'fail',
f'Suspicious entries found: {found_suspicious[:3]}')
add('Security Scan', 'fail', f'Suspicious entries found: {found_suspicious[:3]}')
else:
add('Security Scan', 'pass', 'No path traversal or dangerous entries detected', more=[
'Member paths are checked for .. segments, absolute roots, and sensitive paths '
'(e.g. .ssh, /etc/shadow).',
])
SCRIPT_EXTENSIONS = ('.sh', '.py', '.pl', '.rb', '.bash', '.zsh')
SAFE_PREFIXES = (
'compose-files/',
'volumes/',
'container-configs/',
'configs/',
)
_SCRIPT_EXTENSIONS = ('.sh', '.py', '.pl', '.rb', '.bash', '.zsh')
_SAFE_PREFIXES = ('compose-files/', 'volumes/', 'container-configs/', 'configs/')
def _is_unexpected_executable_script(member):
name = member.name
if any(name.startswith(p) or f'/{p}' in name for p in _SAFE_PREFIXES):
return False
name_lower = name.lower()
has_script_ext = any(name_lower.endswith(ext) for ext in _SCRIPT_EXTENSIONS)
has_exec_bit = bool(member.mode & 0o111)
return has_script_ext and has_exec_bit
def _check_executable_scripts(archive_path, add):
suspicious_scripts = []
try:
with tarfile.open(archive_path, 'r:gz') as tf:
for member in tf.getmembers():
if not member.isfile():
continue
name = member.name
if any(name.startswith(p) or f'/{p}' in name for p in SAFE_PREFIXES):
continue
name_lower = name.lower()
has_script_ext = any(name_lower.endswith(ext) for ext in SCRIPT_EXTENSIONS)
has_exec_bit = bool(member.mode & 0o111)
if has_script_ext and has_exec_bit:
suspicious_scripts.append(os.path.basename(name))
if member.isfile() and _is_unexpected_executable_script(member):
suspicious_scripts.append(os.path.basename(member.name))
except Exception:
pass
@@ -292,6 +293,8 @@ def audit_backup(backup_file, source='local'):
else:
add('Executable Scripts', 'pass', 'No unexpected executable scripts found')
def _check_volume_count(members, add):
vol_archives = [m for m in members if 'volumes/' in m and m.endswith('.tar.gz')]
v = len(vol_archives)
if v == 0:
@@ -301,48 +304,89 @@ def audit_backup(backup_file, source='local'):
else:
add('Volume Count', 'pass', f'{v} volume archives present')
def _score_checks(checks):
weights = {'pass': 10, 'warn': 5, 'fail': 0}
total = len(checks) * 10
earned = sum(weights.get(c['status'], 0) for c in checks)
score = int((earned / total) * 100) if total > 0 else 0
return int((earned / total) * 100) if total > 0 else 0
has_fails = any(c['status'] == 'fail' for c in checks)
ok = not has_fails and score >= 60
def _summary_for_score(score):
if score >= 90:
summary = 'Backup looks healthy and is safe to restore.'
elif score >= 70:
summary = 'Minor warnings — likely safe, but review before restoring.'
elif score >= 40:
summary = 'Significant issues detected — restore with caution.'
else:
summary = 'Multiple checks failed — do NOT restore without manual inspection.'
return 'Backup looks healthy and is safe to restore.'
if score >= 70:
return 'Minor warnings — likely safe, but review before restoring.'
if score >= 40:
return 'Significant issues detected — restore with caution.'
return 'Multiple checks failed — do NOT restore without manual inspection.'
has_warns = any(c['status'] == 'warn' for c in checks)
def _health_tier_for(score, has_fails, has_warns):
if has_fails:
health_tier = 'critical'
health_label = 'Unhealthy'
elif score == 100:
health_tier = 'excellent'
health_label = '100% healthy'
elif score >= 90:
health_tier = 'good'
health_label = 'Healthy' if not has_warns else 'Healthy (with notes)'
elif score >= 70:
health_tier = 'fair'
health_label = 'Mostly healthy'
elif score >= 40:
health_tier = 'poor'
health_label = 'At risk'
else:
health_tier = 'critical'
health_label = 'Unhealthy'
return 'critical', 'Unhealthy'
if score == 100:
return 'excellent', '100% healthy'
if score >= 90:
return 'good', ('Healthy' if not has_warns else 'Healthy (with notes)')
if score >= 70:
return 'fair', 'Mostly healthy'
if score >= 40:
return 'poor', 'At risk'
return 'critical', 'Unhealthy'
def _missing_result(backup_file, checks, detail=None):
return {
'ok': False, 'score': 0, 'checks': checks,
'backup_file': backup_file,
'file_size_bytes': None,
'file_size_display': None,
'health_tier': 'critical',
'health_label': 'Unhealthy',
'summary': detail or 'Cannot access backup file from this host.',
}
def audit_backup(backup_file, source='local'):
checks = []
def add(name, status, detail='', more=None):
entry = {'name': name, 'status': status, 'detail': detail}
if more:
entry['more'] = more
checks.append(entry)
archive_path = _resolve_archive_path(backup_file, source, add)
if archive_path is None:
return _missing_result(backup_file, checks, 'Cannot access backup file from this host.')
if not os.path.exists(archive_path):
add('File Exists', 'fail', f'Not found: {archive_path}')
return _missing_result(backup_file, checks, 'Backup file does not exist on disk.')
add('File Exists', 'pass', archive_path)
size_bytes, size_human = _check_file_size(archive_path, add)
_check_checksum(archive_path, add)
_check_archive_integrity(archive_path, add)
members = _list_archive_members(archive_path)
_check_internal_structure(members, add)
_check_security_scan(members, add)
_check_executable_scripts(archive_path, add)
_check_volume_count(members, add)
score = _score_checks(checks)
has_fails = any(c['status'] == 'fail' for c in checks)
has_warns = any(c['status'] == 'warn' for c in checks)
ok = not has_fails and score >= 60
health_tier, health_label = _health_tier_for(score, has_fails, has_warns)
return {
'ok': ok,
'score': score,
'checks': checks,
'summary': summary,
'summary': _summary_for_score(score),
'backup_file': backup_file,
'file_size_bytes': size_bytes,
'file_size_display': size_human,
@@ -754,7 +798,7 @@ def _get_system_info_fallback():
disk_total_gb = disk.total / (1024**3)
info['disk'] = f"{disk_used_gb:.0f}G/{disk_total_gb:.0f}G"
info['disk_pct'] = str(int(disk.percent))
except:
except Exception:
import shutil
disk = shutil.disk_usage('/')
info['disk'] = f"{disk.used // (1024**3)}G/{disk.total // (1024**3)}G"
@@ -764,7 +808,7 @@ def _get_system_info_fallback():
try:
docker_v = subprocess.check_output(['docker', '--version'], stderr=subprocess.DEVNULL, text=True)
info['docker_v'] = docker_v.split()[-1].strip(',')
except:
except Exception:
pass
except Exception as e: