Python으로 SCC 기반 데이터 이전을 관리하고 이전 영향 평가를 추적하는 시스템입니다.
from dataclasses import dataclass, field
from datetime import date, datetime
from enum import Enum
from typing import Optional
import hashlib
import json
class SCCModule(Enum):
"""SCC 모듈 유형 (2021/914)"""
MODULE_1 = "controller_to_controller"
MODULE_2 = "controller_to_processor"
MODULE_3 = "processor_to_processor"
MODULE_4 = "processor_to_controller"
class TransferRiskLevel(Enum):
"""이전 영향 평가 위험 수준"""
LOW = "low" # 적정성 결정국 수준
MEDIUM = "medium" # 보충적 조치로 해결 가능
HIGH = "high" # 보충적 조치로도 불충분할 수 있음
PROHIBITIVE = "prohibitive" # 이전 불가
class SupplementaryMeasure(Enum):
"""보충적 조치 유형"""
STRONG_ENCRYPTION = "strong_encryption_in_transit_at_rest"
PSEUDONYMISATION = "pseudonymisation_before_transfer"
SPLIT_PROCESSING = "split_processing_across_jurisdictions"
CONTRACTUAL_COMMITMENTS = "enhanced_contractual_commitments"
GOVERNMENT_ACCESS_NOTIFICATION = "government_access_notification"
AUDIT_RIGHTS = "expanded_audit_rights"
@dataclass
class DataImporter:
"""데이터 수입자 (제3국 소재)"""
name: str
country_code: str
country_name: str
has_adequacy_decision: bool
adequacy_scope: Optional[str] = None # 예: "Data Privacy Framework"
surveillance_law_concerns: list[str] = field(default_factory=list)
@dataclass
class SCCAgreement:
"""SCC 계약"""
agreement_id: str
module: SCCModule
data_exporter: str # EU 소재 기업
data_importer: DataImporter
execution_date: date
data_categories: list[str]
data_subjects: list[str]
transfer_purpose: str
supplementary_measures: list[SupplementaryMeasure] = field(default_factory=list)
tia_completed: bool = False
tia_date: Optional[date] = None
tia_risk_level: Optional[TransferRiskLevel] = None
valid: bool = True
def get_agreement_hash(self) -> str:
"""계약 식별 해시"""
content = f"{self.agreement_id}|{self.data_exporter}|{self.data_importer.name}"
return hashlib.sha256(content.encode()).hexdigest()[:12]
@dataclass
class TransferImpactAssessment:
"""이전 영향 평가 (TIA)"""
tia_id: str
scc_agreement_id: str
assessment_date: date
destination_country: str
laws_assessed: list[str]
access_requests_history: int # 최근 정부 접근 요청 수
legal_remedies_available: bool
effective_supplementary_measures: list[SupplementaryMeasure]
risk_level: TransferRiskLevel
assessor: str
next_review_date: date
notes: str = ""
class SCCTransferManager:
"""SCC 기반 국외이전 관리 시스템"""
# 주요국 위험 프로필 (예시)
COUNTRY_RISK_PROFILES = {
"US": {
"has_adequacy": True,
"adequacy_scope": "Data Privacy Framework (DPF)",
"concerns": ["FISA 702", "EO 12333"],
"dpf_certified_required": True
},
"CN": {
"has_adequacy": False,
"concerns": ["National Security Law", "Data Security Law", "PIPL government access"],
"base_risk": TransferRiskLevel.HIGH
},
"JP": {
"has_adequacy": True,
"adequacy_scope": "Full adequacy",
"concerns": [],
"base_risk": TransferRiskLevel.LOW
},
"KR": {
"has_adequacy": True,
"adequacy_scope": "Full adequacy (2021)",
"concerns": [],
"base_risk": TransferRiskLevel.LOW
},
"IN": {
"has_adequacy": False,
"concerns": ["IT Act surveillance provisions"],
"base_risk": TransferRiskLevel.MEDIUM
}
}
def __init__(self):
self.agreements: dict[str, SCCAgreement] = {}
self.tia_records: dict[str, TransferImpactAssessment] = {}
def determine_scc_module(
self,
exporter_role: str,
importer_role: str
) -> SCCModule:
"""당사자 역할에 따른 SCC 모듈 결정"""
role_map = {
("controller", "controller"): SCCModule.MODULE_1,
("controller", "processor"): SCCModule.MODULE_2,
("processor", "processor"): SCCModule.MODULE_3,
("processor", "controller"): SCCModule.MODULE_4,
}
key = (exporter_role.lower(), importer_role.lower())
if key not in role_map:
raise ValueError(f"Invalid role combination: {exporter_role} -> {importer_role}")
return role_map[key]
def create_scc_agreement(
self,
data_exporter: str,
data_importer: DataImporter,
module: SCCModule,
data_categories: list[str],
transfer_purpose: str
) -> SCCAgreement:
"""SCC 계약 생성"""
agreement = SCCAgreement(
agreement_id=f"SCC-{datetime.now().strftime('%Y%m%d')}-{len(self.agreements)+1:04d}",
module=module,
data_exporter=data_exporter,
data_importer=data_importer,
execution_date=date.today(),
data_categories=data_categories,
data_subjects=["employees", "customers"], # 예시
transfer_purpose=transfer_purpose
)
self.agreements[agreement.agreement_id] = agreement
return agreement
def conduct_transfer_impact_assessment(
self,
agreement_id: str,
laws_assessed: list[str],
access_requests: int,
legal_remedies: bool,
assessor: str
) -> TransferImpactAssessment:
"""이전 영향 평가 (TIA) 수행 - Schrems II 요건"""
agreement = self.agreements.get(agreement_id)
if not agreement:
raise ValueError(f"Agreement {agreement_id} not found")
importer = agreement.data_importer
country_profile = self.COUNTRY_RISK_PROFILES.get(importer.country_code, {})
# 위험 수준 결정
if country_profile.get("has_adequacy"):
# 적정성 결정이 있는 경우
if country_profile.get("dpf_certified_required"):
# US DPF: 인증 여부 확인 필요
risk_level = TransferRiskLevel.LOW # 인증된 경우
else:
risk_level = TransferRiskLevel.LOW
else:
# 적정성 결정이 없는 경우
base_risk = country_profile.get("base_risk", TransferRiskLevel.MEDIUM)
concerns = country_profile.get("concerns", [])
if len(concerns) >= 2 and not legal_remedies:
risk_level = TransferRiskLevel.HIGH
elif access_requests > 10: # 빈번한 정부 접근
risk_level = TransferRiskLevel.HIGH
else:
risk_level = base_risk
# 필요한 보충적 조치 결정
supplementary = self._determine_supplementary_measures(risk_level, country_profile)
tia = TransferImpactAssessment(
tia_id=f"TIA-{agreement_id}",
scc_agreement_id=agreement_id,
assessment_date=date.today(),
destination_country=importer.country_code,
laws_assessed=laws_assessed,
access_requests_history=access_requests,
legal_remedies_available=legal_remedies,
effective_supplementary_measures=supplementary,
risk_level=risk_level,
assessor=assessor,
next_review_date=date(date.today().year + 1, date.today().month, date.today().day)
)
# 계약에 TIA 정보 업데이트
agreement.tia_completed = True
agreement.tia_date = date.today()
agreement.tia_risk_level = risk_level
agreement.supplementary_measures = supplementary
self.tia_records[tia.tia_id] = tia
return tia
def _determine_supplementary_measures(
self,
risk_level: TransferRiskLevel,
country_profile: dict
) -> list[SupplementaryMeasure]:
"""위험 수준에 따른 보충적 조치 결정"""
measures = []
if risk_level in [TransferRiskLevel.MEDIUM, TransferRiskLevel.HIGH]:
measures.append(SupplementaryMeasure.STRONG_ENCRYPTION)
if risk_level == TransferRiskLevel.HIGH:
measures.append(SupplementaryMeasure.PSEUDONYMISATION)
measures.append(SupplementaryMeasure.GOVERNMENT_ACCESS_NOTIFICATION)
measures.append(SupplementaryMeasure.AUDIT_RIGHTS)
measures.append(SupplementaryMeasure.CONTRACTUAL_COMMITMENTS)
return measures
def validate_transfer(self, agreement_id: str, data_record: dict) -> dict:
"""실제 데이터 이전 검증"""
agreement = self.agreements.get(agreement_id)
if not agreement:
return {"allowed": False, "reason": "SCC agreement not found"}
if not agreement.valid:
return {"allowed": False, "reason": "SCC agreement no longer valid"}
if not agreement.tia_completed:
return {"allowed": False, "reason": "Transfer Impact Assessment not completed"}
if agreement.tia_risk_level == TransferRiskLevel.PROHIBITIVE:
return {"allowed": False, "reason": "Transfer prohibited due to high risk"}
# 보충적 조치 적용 확인
if agreement.tia_risk_level == TransferRiskLevel.HIGH:
if SupplementaryMeasure.PSEUDONYMISATION in agreement.supplementary_measures:
if not self._is_pseudonymised(data_record):
return {
"allowed": False,
"reason": "Data must be pseudonymised before transfer"
}
return {
"allowed": True,
"agreement_id": agreement_id,
"module": agreement.module.value,
"destination": agreement.data_importer.country_code,
"supplementary_measures_applied": [m.value for m in agreement.supplementary_measures]
}
def _is_pseudonymised(self, data: dict) -> bool:
"""데이터 가명처리 여부 확인"""
direct_identifiers = ["name", "email", "phone", "ssn", "address"]
return not any(key in data for key in direct_identifiers)
def get_transfer_summary(self) -> dict:
"""전체 이전 현황 요약"""
by_country = {}
by_module = {}
by_risk = {}
for agreement in self.agreements.values():
country = agreement.data_importer.country_code
by_country[country] = by_country.get(country, 0) + 1
module = agreement.module.value
by_module[module] = by_module.get(module, 0) + 1
if agreement.tia_risk_level:
risk = agreement.tia_risk_level.value
by_risk[risk] = by_risk.get(risk, 0) + 1
return {
"total_agreements": len(self.agreements),
"by_destination_country": by_country,
"by_scc_module": by_module,
"by_risk_level": by_risk,
"tia_completion_rate": f"{sum(1 for a in self.agreements.values() if a.tia_completed)/len(self.agreements)*100:.1f}%"
}
# 사용 예시
if __name__ == "__main__":
manager = SCCTransferManager()
# 데이터 수입자 정의 (미국 클라우드 서비스)
us_processor = DataImporter(
name="CloudAI Inc.",
country_code="US",
country_name="United States",
has_adequacy_decision=True,
adequacy_scope="Data Privacy Framework",
surveillance_law_concerns=["FISA 702"]
)
# SCC 계약 생성
module = manager.determine_scc_module("controller", "processor")
agreement = manager.create_scc_agreement(
data_exporter="EU Tech GmbH",
data_importer=us_processor,
module=module,
data_categories=["customer_profiles", "usage_data"],
transfer_purpose="AI model training and inference"
)
# TIA 수행
tia = manager.conduct_transfer_impact_assessment(
agreement_id=agreement.agreement_id,
laws_assessed=["FISA 702", "EO 12333", "CLOUD Act"],
access_requests=2,
legal_remedies=True,
assessor="DPO Team"
)
print(f"위험 수준: {tia.risk_level.value}")
print(f"보충적 조치: {[m.value for m in tia.effective_supplementary_measures]}")