고급 보안 엔지니어 면접 질문: 완벽 가이드

# 예시: 안전한 API 설계
from functools import wraps
from flask import request, abort
import jwt

def require_auth(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        token = request.headers.get('Authorization')
        if not token:
            abort(401)
        
        try:
            payload = jwt.decode(token, SECRET_KEY, algorithms=['HS256'])
            # 속도 제한
            if check_rate_limit(payload['user_id']):
                abort(429)
            # 권한 부여
            if not has_permission(payload['user_id'], request.path):
                abort(403)
        except jwt.InvalidTokenError:
            abort(401)
        
        return f(*args, **kwargs)
    return decorated

@app.route('/api/sensitive-data')
@require_auth
def get_sensitive_data():
    # 감사 로깅
    log_access(request.user_id, 'sensitive-data', 'read')
    return jsonify(data)

# CSPM 자동화 예시
import boto3
from datetime import datetime

class AWSSecurityPosture:
    def __init__(self):
        self.ec2 = boto3.client('ec2')
        self.s3 = boto3.client('s3')
        self.iam = boto3.client('iam')
        self.findings = []
    
    def check_s3_encryption(self):
        """S3 버킷 암호화 확인"""
        buckets = self.s3.list_buckets()['Buckets']
        
        for bucket in buckets:
            bucket_name = bucket['Name']
            try:
                encryption = self.s3.get_bucket_encryption(
                    Bucket=bucket_name
                )
            except:
                self.findings.append({
                    'severity': 'high',
                    'resource': bucket_name,
                    'issue': 'S3 버킷이 암호화되지 않았습니다.',
                    'remediation': '기본 암호화 활성화'
                })
    
    def check_security_groups(self):
        """지나치게 허용적인 보안 그룹 확인"""
        sgs = self.ec2.describe_security_groups()['SecurityGroups']
        
        for sg in sgs:
            for rule in sg.get('IpPermissions', []):
                for ip_range in rule.get('IpRanges', []):
                    if ip_range.get('CidrIp') == '0.0.0.0/0':
                        self.findings.append({
                            'severity': 'critical',
                            'resource': sg['GroupId'],
                            'issue': f"보안 그룹이 포트 {rule.get('FromPort')}에서 0.0.0.0/0을 허용합니다.",
                            'remediation': '소스 IP 범위 제한'
                        })
    
    def check_iam_policies(self):
        """지나치게 허용적인 IAM 정책 확인"""
        users = self.iam.list_users()['Users']
        
        for user in users:
            policies = self.iam.list_attached_user_policies(
                UserName=user['UserName']
            )['AttachedPolicies']
            
            for policy in policies:
                if policy['PolicyName'] == 'AdministratorAccess':
                    self.findings.append({
                        'severity': 'high',
                        'resource': user['UserName'],
                        'issue': '사용자에게 AdministratorAccess 권한이 있습니다.',
                        'remediation': '최소 권한 원칙 적용'
                    })
    
    def generate_report(self):
        self.check_s3_encryption()
        self.check_security_groups()
        self.check_iam_policies()
        
        return {
            'timestamp': datetime.now().isoformat(),
            'total_findings': len(self.findings),
            'critical': len([f for f in self.findings if f['severity'] == 'critical']),
            'high': len([f for f in self.findings if f['severity'] == 'high']),
            'findings': self.findings
        }

# Pod 보안 정책
apiVersion: policy/v1beta1
kind: PodSecurityPolicy
metadata:
  name: restricted
spec:
  privileged: false
  allowPrivilegeEscalation: false
  requiredDropCapabilities:
    - ALL
  volumes:
    - 'configMap'
    - 'emptyDir'
    - 'projected'
    - 'secret'
  hostNetwork: false
  hostIPC: false
  hostPID: false
  runAsUser:
    rule: 'MustRunAsNonRoot'
  seLinux:
    rule: 'RunAsAny'
  fsGroup:
    rule: 'RunAsAny'
  readOnlyRootFilesystem: true

# Pod 격리를 위한 네트워크 정책
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: deny-all-ingress
  namespace: production
spec:
  podSelector: {}
  policyTypes:
  - Ingress
  - Egress
  ingress: []
  egress:
  - to:
    - namespaceSelector:
        matchLabels:
          name: kube-system
    ports:
    - protocol: UDP
      port: 53

# 안전한 Lambda 함수 예시
import json
import boto3
import os
from aws_xray_sdk.core import xray_recorder
from aws_lambda_powertools import Logger, Tracer

logger = Logger()
tracer = Tracer()

@tracer.capture_lambda_handler
@logger.inject_lambda_context
def lambda_handler(event, context):
    # 입력 유효성 검사
    if 'user_id' not in event:
        return {
            'statusCode': 400,
            'body': json.dumps({'error': 'Missing user_id'})
        }
    
    # 하드 코딩된 자격 증명 대신 IAM 역할 사용
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table(os.environ['TABLE_NAME'])
    
    # 최소 권한 - 필요한 것만 접근
    try:
        response = table.get_item(
            Key={'user_id': event['user_id']}
        )
        
        # 내부 오류 노출 금지
        return {
            'statusCode': 200,
            'body': json.dumps(response.get('Item', {}))
        }
    except Exception as e:
        logger.error(f"Error: {str(e)}")
        return {
            'statusCode': 500,
            'body': json.dumps({'error': 'Internal server error'})
        }

# GitLab CI/CD 보안 파이프라인
stages:
  - build
  - test
  - security
  - deploy

# SAST 스캔
sast:
  stage: security
  script:
    - sonar-scanner \
        -Dsonar.projectKey=myapp \
        -Dsonar.sources=. \
        -Dsonar.host.url=$SONAR_URL
  allow_failure: false

# 의존성 스캔
dependency_scan:
  stage: security
  script:
    - npm audit --audit-level=high
    - snyk test --severity-threshold=high
  allow_failure: false

# 컨테이너 스캔
container_scan:
  stage: security
  script:
    - trivy image --severity HIGH,CRITICAL myapp:latest
  allow_failure: false

# DAST 스캔
dast:
  stage: security
  script:
    - zap-baseline.py -t https://staging.example.com
  allow_failure: true

# 자동화된 보안 검사
class SecurityCodeReview:
    def __init__(self, code_diff):
        self.code_diff = code_diff
        self.issues = []
    
    def check_hardcoded_secrets(self):
        """하드 코딩된 자격 증명 감지"""
        patterns = [
            r'password\s*=\s*["\'].*["\']',
            r'api_key\s*=\s*["\'].*["\']',
            r'secret\s*=\s*["\'].*["\']',
            r'token\s*=\s*["\'].*["\']'
        ]
        
        for pattern in patterns:
            if re.search(pattern, self.code_diff, re.IGNORECASE):
                self.issues.append({
                    'type': 'hardcoded_secret',
                    'severity': 'critical',
                    'message': '잠재적인 하드 코딩된 자격 증명이 감지되었습니다.'
                })
    
    def check_sql_injection(self):
        """잠재적인 SQL 인젝션 감지"""
        if re.search(r'execute\([^?].*\+.*\)', self.code_diff):
            self.issues.append({
                'type': 'sql_injection',
                'severity': 'high',
                'message': '잠재적인 SQL 인젝션 - 매개변수화된 쿼리를 사용하십시오.'
            })
    
    def check_xss(self):
        """잠재적인 XSS 감지"""
        if 'innerHTML' in self.code_diff or 'dangerouslySetInnerHTML' in self.code_diff:
            self.issues.append({
                'type': 'xss',
                'severity': 'high',
                'message': '잠재적인 XSS - 사용자 입력을 삭제하십시오.'
            })

# STRIDE 위협 모델
threat_model = {
    'application': 'Payment API',
    'threats': [
        {
            'category': 'Spoofing',
            'threat': '공격자가 합법적인 사용자를 사칭합니다.',
            'mitigation': 'OAuth 2.0 + JWT 구현',
            'status': 'implemented'
        },
        {
            'category': 'Tampering',
            'threat': '전송 중 트랜잭션 금액이 수정되었습니다.',
            'mitigation': 'HTTPS + 요청 서명 사용',
            'status': 'implemented'
        },
        {
            'category': 'Repudiation',
            'threat': '사용자가 트랜잭션 수행을 부인합니다.',
            'mitigation': '포괄적인 감사 로깅',
            'status': 'planned'
        },
        {
            'category': 'Information Disclosure',
            'threat': '로그에 민감한 데이터가 노출되었습니다.',
            'mitigation': '로그 삭제, PII 암호화',
            'status': 'in_progress'
        },
        {
            'category': 'Denial of Service',
            'threat': 'API가 요청으로 압도되었습니다.',
            'mitigation': '속도 제한 + WAF',
            'status': 'implemented'
        },
        {
            'category': 'Elevation of Privilege',
            'threat': '사용자가 관리자 기능에 접근합니다.',
            'mitigation': 'RBAC + 권한 부여 검사',
            'status': 'implemented'
        }
    ]
}

# 자동화된 사고 대응 스크립트
import subprocess
import datetime

def isolate_host(ip_address):
    """손상된 호스트 격리"""
    # 방화벽에서 차단
    subprocess.run([
        'iptables', '-A', 'INPUT', '-s', ip_address, '-j', 'DROP'
    ])
    
    # 호스트에서 네트워크 인터페이스 비활성화
    subprocess.run([
        'ssh', f'admin@{ip_address}',
        'sudo', 'ifconfig', 'eth0', 'down'
    ])
    
    # 작업 기록
    log_incident_action(
        action='host_isolation',
        target=ip_address,
        timestamp=datetime.datetime.now()
    )

def collect_forensics(ip_address):
    """포렌식 데이터 수집"""
    commands = [
        'netstat -an',  # 네트워크 연결
        'ps aux',       # 실행 중인 프로세스
        'last -a',      # 로그인 기록
        'find / -mtime -1',  # 최근 수정된 파일
    ]
    
    for cmd in commands:
        output = subprocess.run(
            ['ssh', f'admin@{ip_address}', cmd],
            capture_output=True
        )
        save_evidence(cmd, output.stdout)

# 지속적인 인증
def verify_access(user, resource, context):
    # 다단계 인증
    if not verify_mfa(user):
        return False
    
    # 장치 상태 확인
    if not is_device_compliant(context['device_id']):
        return False
    
    # 위치 확인
    if context['ip'] not in ALLOWED_LOCATIONS:
        return False
    
    # 위험 점수
    risk_score = calculate_risk(user, context)
    if risk_score > THRESHOLD:
        require_step_up_auth()
    
    # Just-in-time 접근
    grant_temporary_access(user, resource, duration='1h')
    
    return True

# 네트워크 정책 (Kubernetes)
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: api-policy
spec:
  podSelector:
    matchLabels:
      app: api
  policyTypes:
  - Ingress
  - Egress
  ingress:
  - from:
    - podSelector:
        matchLabels:
          app: frontend
    ports:
    - protocol: TCP
      port: 8080
  egress:
  - to:
    - podSelector:
        matchLabels:
          app: database
    ports:
    - protocol: TCP
      port: 5432

# 규정 준수 평가 도구
def assess_compliance(framework='NIST'):
    controls = load_framework_controls(framework)
    results = []
    
    for control in controls:
        status = check_control_implementation(control)
        results.append({
            'control_id': control.id,
            'description': control.description,
            'status': status,  # Implemented, Partial, Not Implemented
            'evidence': collect_evidence(control),
            'remediation': get_remediation_plan(control) if status != 'Implemented' else None
        })
    
    return generate_compliance_report(results)

# 위협 인텔리전스 자동화
import requests

def check_ioc(indicator, ioc_type='ip'):
    """침해 지표 확인"""
    # 위협 인텔리전스 플랫폼 쿼리
    response = requests.get(
        f'https://api.threatintel.com/v1/indicators/{ioc_type}/{indicator}',
        headers={'Authorization': f'Bearer {API_KEY}'}
    )
    
    if response.json()['malicious']:
        # 방화벽에서 차단
        block_indicator(indicator, ioc_type)
        
        # SOC에 경고
        create_alert(
            severity='high',
            title=f'악성 {ioc_type} 감지',
            description=f'{indicator}가 악성으로 식별되었습니다.',
            ioc=indicator,
            threat_actor=response.json()['threat_actor']
        )
        
        # 관련 활동 탐색
        hunt_for_ioc(indicator)

def enrich_alert(alert):
    """위협 인텔리전스로 보안 경고 보강"""
    context = {
        'ip_reputation': get_ip_reputation(alert.source_ip),
        'domain_age': get_domain_age(alert.domain),
        'ssl_cert': analyze_ssl_cert(alert.domain),
        'similar_incidents': find_similar_incidents(alert)
    }
    
    alert.context = context
    alert.risk_score = calculate_risk_score(context)
    return alert

# ATT&CK 기술 기반 헌팅
class ThreatHunt:
    def __init__(self, siem_client):
        self.siem = siem_client
        self.findings = []
    
    def hunt_t1003_credential_dumping(self):
        """자격 증명 덤핑 (T1003) 헌팅"""
        query = '''
        EventID:4688 AND 
        (ProcessName:*lsass.exe* OR 
         ProcessName:*procdump* OR 
         ProcessName:*mimikatz*)
        '''
        
        results = self.siem.search(query, timeframe='24h')
        
        for event in results:
            self.findings.append({
                'technique': 'T1003 - Credential Dumping',
                'host': event['hostname'],
                'process': event['process_name'],
                'user': event['username'],
                'timestamp': event['timestamp'],
                'severity': 'critical'
            })
    
    def hunt_t1078_valid_accounts(self):
        """의심스러운 계정 사용 (T1078) 헌팅"""
        query = '''
        EventID:4624 AND 
        LogonType:3 AND
        SourceIP:* NOT (SourceIP:10.* OR SourceIP:192.168.*)
        '''
        
        # 비정상적인 로그인 패턴 검색
        results = self.siem.search(query, timeframe='7d')
        
        # 이상 징후 감지
        user_logins = {}
        for event in results:
            user = event['username']
            if user not in user_logins:
                user_logins[user] = []
            user_logins[user].append(event)
        
        # 여러 국가에서 로그인한 사용자 플래그
        for user, logins in user_logins.items():
            countries = set([l['source_country'] for l in logins])
            if len(countries) > 2:
                self.findings.append({
                    'technique': 'T1078 - Valid Accounts',
                    'user': user,
                    'countries': list(countries),
                    'severity': 'high'
                })
    
    def hunt_t1059_command_execution(self):
        """의심스러운 명령 실행 (T1059) 헌팅"""
        suspicious_commands = [
            'powershell -enc',
            'powershell -e ',
            'cmd /c echo',
            'wscript',
            'cscript'
        ]
        
        for cmd in suspicious_commands:
            query = f'EventID:4688 AND CommandLine:*{cmd}*'
            results = self.siem.search(query, timeframe='24h')
            
            for event in results:
                self.findings.append({
                    'technique': 'T1059 - Command Execution',
                    'host': event['hostname'],
                    'command': event['command_line'],
                    'user': event['username'],
                    'severity': 'high'
                })

-- 수평 이동을 위한 Splunk 쿼리
index=windows EventCode=4624 LogonType=3
| stats dc(ComputerName) as unique_hosts by Account_Name
| where unique_hosts > 10
| sort - unique_hosts

-- 대규모 아웃바운드 전송
index=firewall action=allowed
| stats sum(bytes_out) as total_bytes by src_ip, dest_ip
| where total_bytes > 1000000000
| sort - total_bytes

-- 레지스트리 Run 키
index=windows EventCode=13 
TargetObject="*\\CurrentVersion\\Run*"
| table _time, ComputerName, User, TargetObject, Details

# 자동화된 헌팅 일정
import schedule
import time

def run_daily_hunts():
    hunter = ThreatHunt(siem_client)
    
    # 모든 헌트 쿼리 실행
    hunter.hunt_t1003_credential_dumping()
    hunter.hunt_t1078_valid_accounts()
    hunter.hunt_t1059_command_execution()
    
    # 보고서 생성
    if hunter.findings:
        report = generate_hunt_report(hunter.findings)
        send_to_slack(report)
        create_tickets(hunter.findings)

# 매일 새벽 2시에 예약
schedule.every().day.at("02:00").do(run_daily_hunts)

while True:
    schedule.run_pending()
    time.sleep(60)

# 위협 헌트 보고서

**날짜:** 2024-01-15
**헌터:** 보안 팀
**가설:** 자격 증명 덤핑 활동

## 발견 사항
- LSASS 메모리 접근 3건 감지
- 모두 동일한 손상된 호스트에서 발생: WS-1234
- 사용자: contractor_account

## 조치 사항
1. 영향을 받는 호스트 격리
2. contractor_account에 대한 자격 증명 재설정
3. 포렌식 이미지 수집
4. 향후 사고에 대한 탐지 규칙 생성

## 배운 점
- 계약자 계정에 대한 더 나은 모니터링 필요
- 권한 있는 작업에 대한 JIT 접근 구현

# 보안 오케스트레이션 플레이북
def handle_phishing_alert(alert):
    """자동화된 피싱 대응"""
    # 지표 추출
    email_data = parse_email(alert.email_id)
    sender = email_data['from']
    links = extract_links(email_data['body'])
    attachments = email_data['attachments']
    
    # 위협 분석
    for link in links:
        if is_malicious_url(link):
            block_url(link)
    
    for attachment in attachments:
        if is_malicious_file(attachment):
            quarantine_file(attachment)
    
    # 보낸 사람 차단
    if sender_reputation(sender) < THRESHOLD:
        block_sender(sender)
    
    # 유사한 이메일 검색
    similar_emails = search_emails(
        sender=sender,
        timeframe='24h'
    )
    
    # 모든 메일함에서 삭제
    for email in similar_emails:
        delete_email(email.id)
        notify_recipient(email.recipient)
    
    # 위협 인텔리전스 업데이트
    add_to_blocklist(sender, links)
    
    # 티켓 생성
    create_incident_ticket(
        title=f'{sender}로부터의 피싱 캠페인',
        affected_users=len(similar_emails)
    )