BigQuery 마스킹된 로그 쿼리 가이드

개요

DTx 플랫폼의 모든 로그는 Google Cloud BigQuery에 저장되며, 현재 부분적으로 마스킹이 적용되고 있습니다. 이 문서는 BigQuery에서 마스킹된 로그를 조회하고 분석하는 방법을 안내합니다.

1. BigQuery 로그 테이블 구조

1.1 데이터셋 및 테이블 구성

데이터셋: dta-cloud-de-dev.log

테이블명	설명	마스킹 대상 필드	보존 기간
api_requests	HTTP API 요청 로그	user_id, client_ip, request_headers, response_body	30일
system_logs	일반 시스템 로그	user_id, metadata	30일
error_logs	에러 및 예외 로그	user_id, metadata	90일
audit	감사 로그	user_id, details	365일
mobile_error_logs	모바일 에러 로그	user_id, user_cycle_id	90일
mobile_info_logs	모바일 정보 로그	user_id, user_cycle_id	30일
performance_metrics	성능 메트릭	user_id, correlation_id	90일

1.2 현재 마스킹 적용 현황

✅ 마스킹 적용된 필드

• password → *****
• token → *****
• passwordHash → *****
• email → bo***@weltcorp.com (부분 마스킹)
• client_ip → 192.168.1.*** (부분 마스킹)
• verificationId → bd54-****-****-64fb (개인 인증 ID 부분 마스킹) 🆕
• sessionId → sess-****-****-1234 (개인 세션 ID 부분 마스킹) 🆕
• userId → user-****-****-5678 (사용자 ID 부분 마스킹) 🆕
• authorization 헤더 → [REDACTED]
• cookie 헤더 → [REDACTED]
• queryParams → 민감 필드 자동 마스킹

⚠️ 의도적으로 마스킹하지 않는 필드들 (운영상 필수)

• correlationId - 분산 시스템에서 요청 추적용 (오류 추적에 필수)
• requestId - HTTP 요청 추적용 (성능 모니터링에 필수)
• traceId - APM 분산 추적용 (장애 대응에 필수)

중요: 이 필드들은 개인정보가 아니며, 운영팀의 장애 대응과 시스템 모니터링에 필수적입니다.

❌ 마스킹 누락된 필드

• user_id - 개인 식별자 (해시화 필요)
• user_agent - 사용자 에이전트

🎯 실무적 마스킹 원칙

API 응답 vs 로그 분리:

• API 응답: 권한 기반 차등 마스킹 (본인 조회 시 원본, 관리자 조회 시 부분 마스킹)
• 로그: 항상 마스킹/해시화 (운영 추적은 가능하되 개인정보 보호)

IP 주소 처리:

• ❌ 완전 해시화 (역추적 불가능)
• ✅ 부분 마스킹 + 지역/ISP 메타데이터 (보안 분석 가능)

2. 마스킹된 로그 조회 쿼리

2.1 API 요청 로그에서 마스킹 상태 확인

-- 최근 1시간 API 요청 로그에서 마스킹 상태 확인
SELECT 
  timestamp,
  request_id,
  method,
  path,
  user_id,  -- 이 필드가 마스킹되어야 함
  client_ip, -- 이 필드가 마스킹되어야 함
  status_code,
  response_time,
  -- 헤더에서 민감 정보 확인
  JSON_EXTRACT_SCALAR(request_headers, '$.authorization') as auth_header,
  JSON_EXTRACT_SCALAR(request_headers, '$.cookie') as cookie_header,
  -- 응답 바디에서 민감 정보 확인
  JSON_EXTRACT_SCALAR(response_body, '$.password') as password_field,
  JSON_EXTRACT_SCALAR(response_body, '$.token') as token_field
FROM `dta-cloud-de-dev.log.api_requests`
WHERE timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 HOUR)
  AND deploy_env = 'dev'
ORDER BY timestamp DESC
LIMIT 10;

2.2 로그인 요청에서 비밀번호 마스킹 확인

-- 로그인 API 요청에서 비밀번호 마스킹 상태 확인
SELECT 
  timestamp,
  request_id,
  path,
  -- metadata에서 요청 바디 확인 (마스킹 상태)
  JSON_EXTRACT_SCALAR(metadata, '$.requestBody.password') as masked_password,
  JSON_EXTRACT_SCALAR(metadata, '$.requestBody.email') as request_email,
  -- 응답에서 토큰 마스킹 확인
  JSON_EXTRACT_SCALAR(response_body, '$.accessToken') as masked_token,
  status_code
FROM `dta-cloud-de-dev.log.api_requests`
WHERE path LIKE '%/auth/login%'
  AND timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 24 HOUR)
ORDER BY timestamp DESC
LIMIT 5;

2.3 사용자별 로그 패턴 분석 (마스킹 고려)

-- 사용자 ID가 노출되는 로그 패턴 분석
SELECT 
  DATE(timestamp) as log_date,
  service,
  COUNT(*) as log_count,
  COUNT(DISTINCT user_id) as unique_users,
  -- 마스킹되지 않은 사용자 ID 패턴 확인
  ARRAY_AGG(DISTINCT 
    CASE 
      WHEN user_id IS NOT NULL AND user_id != '[MASKED]' 
      THEN CONCAT(SUBSTR(user_id, 1, 8), '-****') 
    END 
    IGNORE NULLS 
    LIMIT 5
  ) as sample_user_ids
FROM `dta-cloud-de-dev.log.system_logs`
WHERE timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY)
  AND user_id IS NOT NULL
GROUP BY log_date, service
ORDER BY log_date DESC, log_count DESC;

2.4 에러 로그에서 민감 정보 노출 확인

-- 에러 로그에서 민감 정보가 스택 트레이스에 포함되었는지 확인
SELECT 
  timestamp,
  service,
  error_code,
  message,
  -- 스택 트레이스에서 민감 정보 패턴 검색
  CASE 
    WHEN REGEXP_CONTAINS(stack_trace, r'password|token|secret|key') THEN 'SENSITIVE_DATA_FOUND'
    ELSE 'CLEAN'
  END as sensitive_data_status,
  -- 사용자 ID 마스킹 상태
  CASE 
    WHEN user_id IS NOT NULL AND user_id != '[MASKED]' THEN 'EXPOSED'
    ELSE 'MASKED'
  END as user_id_status
FROM `dta-cloud-de-dev.log.error_logs`
WHERE timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 24 HOUR)
  AND (
    REGEXP_CONTAINS(stack_trace, r'password|token|secret|key')
    OR user_id IS NOT NULL
  )
ORDER BY timestamp DESC
LIMIT 20;

2.5 IP 주소 및 지역 정보 분석 (실무적 접근)

-- 현재 스키마: IP 주소 완전 노출 확인
SELECT 
  DATE(timestamp) as log_date,
  -- 현재 IP 노출 상태 확인
  CASE 
    WHEN client_ip IS NULL THEN 'NO_IP'
    WHEN client_ip = '[REDACTED]' THEN 'MASKED'
    WHEN REGEXP_CONTAINS(client_ip, r'^\*+\.\*+\.\*+\.\d+$') THEN 'PARTIALLY_MASKED'
    WHEN REGEXP_CONTAINS(client_ip, r'^\d+\.\d+\.\d+\.\d+$') THEN 'EXPOSED'
    ELSE 'UNKNOWN'
  END as ip_masking_status,
  COUNT(*) as request_count,
  -- 네트워크 대역별 분석 (보안 패턴 탐지용)
  COUNT(DISTINCT REGEXP_EXTRACT(client_ip, r'^(\d+\.\d+\.\d+)\.\d+$')) as unique_network_blocks
FROM `dta-cloud-de-dev.log.api_requests`
WHERE timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY)
GROUP BY log_date, ip_masking_status
ORDER BY log_date DESC;

2.6 개선된 IP 처리 후 예상 쿼리

-- 개선 후: 부분 마스킹 + 메타데이터 기반 보안 분석
SELECT 
  DATE(timestamp) as log_date,
  client_ip_masked,        -- "192.168.1.***"
  client_region,           -- "Seoul, KR"
  client_isp,             -- "KT Corporation"
  COUNT(*) as request_count,
  COUNT(DISTINCT user_hash) as unique_users,
  -- 의심 활동 탐지
  COUNTIF(status_code = 401) as auth_failures,
  COUNTIF(status_code >= 500) as server_errors
FROM `dta-cloud-de-dev.log.api_requests`
WHERE timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 24 HOUR)
  AND client_ip_masked IS NOT NULL
GROUP BY log_date, client_ip_masked, client_region, client_isp
HAVING auth_failures > 10 OR server_errors > 5  -- 의심 활동 임계값
ORDER BY auth_failures DESC, server_errors DESC;

3. 마스킹 효과성 모니터링 쿼리

3.1 마스킹 누락 탐지

-- 마스킹되지 않은 민감 정보 탐지
WITH sensitive_patterns AS (
  SELECT 
    timestamp,
    request_id,
    'EMAIL' as data_type,
    'response_body' as location,
    JSON_EXTRACT_SCALAR(response_body, '$.email') as value
  FROM `dta-cloud-de-dev.log.api_requests`
  WHERE JSON_EXTRACT_SCALAR(response_body, '$.email') IS NOT NULL
    AND NOT REGEXP_CONTAINS(JSON_EXTRACT_SCALAR(response_body, '$.email'), r'^.{2}\*+@')
    AND timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 HOUR)
  
  UNION ALL
  
  SELECT 
    timestamp,
    request_id,
    'PASSWORD' as data_type,
    'request_headers' as location,
    JSON_EXTRACT_SCALAR(request_headers, '$.authorization') as value
  FROM `dta-cloud-de-dev.log.api_requests`
  WHERE JSON_EXTRACT_SCALAR(request_headers, '$.authorization') IS NOT NULL
    AND JSON_EXTRACT_SCALAR(request_headers, '$.authorization') != '[REDACTED]'
    AND timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 HOUR)
)
SELECT 
  data_type,
  location,
  COUNT(*) as violations,
  ARRAY_AGG(DISTINCT request_id LIMIT 5) as sample_requests
FROM sensitive_patterns
WHERE value IS NOT NULL
GROUP BY data_type, location
ORDER BY violations DESC;

3.2 마스킹 성공률 계산

-- 마스킹 성공률 대시보드
WITH masking_stats AS (
  SELECT 
    DATE(timestamp) as log_date,
    service,
    COUNT(*) as total_logs,
    -- 사용자 ID 마스킹 률
    COUNTIF(user_id IS NULL OR user_id = '[MASKED]') as masked_user_ids,
    -- Authorization 헤더 마스킹률  
    COUNTIF(
      JSON_EXTRACT_SCALAR(request_headers, '$.authorization') IS NULL OR 
      JSON_EXTRACT_SCALAR(request_headers, '$.authorization') = '[REDACTED]'
    ) as masked_auth_headers,
    COUNTIF(JSON_EXTRACT_SCALAR(request_headers, '$.authorization') IS NOT NULL) as total_auth_headers
  FROM `dta-cloud-de-dev.log.api_requests`
  WHERE timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY)
  GROUP BY log_date, service
)
SELECT 
  log_date,
  service,
  total_logs,
  ROUND(masked_user_ids * 100.0 / total_logs, 2) as user_id_masking_rate,
  ROUND(
    CASE 
      WHEN total_auth_headers > 0 THEN masked_auth_headers * 100.0 / total_auth_headers
      ELSE 100
    END, 2
  ) as auth_header_masking_rate
FROM masking_stats
ORDER BY log_date DESC, service;

4. 실시간 마스킹 모니터링

4.1 실시간 민감정보 노출 알림 쿼리

-- 최근 15분간 마스킹 실패한 로그 탐지
SELECT 
  timestamp,
  service,
  method,
  path,
  user_id,
  'USER_ID_EXPOSED' as violation_type,
  CONCAT('User ID exposed: ', SUBSTR(user_id, 1, 8), '...') as description
FROM `dta-cloud-de-dev.log.api_requests`
WHERE timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 15 MINUTE)
  AND user_id IS NOT NULL 
  AND user_id != '[MASKED]'
  AND LENGTH(user_id) > 10  -- UUID 길이 체크

UNION ALL

SELECT 
  timestamp,
  service,
  method,
  path,
  client_ip as user_id,
  'IP_ADDRESS_EXPOSED' as violation_type,
  CONCAT('IP address exposed: ', client_ip) as description
FROM `dta-cloud-de-dev.log.api_requests`
WHERE timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 15 MINUTE)
  AND client_ip IS NOT NULL
  AND client_ip != '[REDACTED]'
  AND REGEXP_CONTAINS(client_ip, r'^\d+\.\d+\.\d+\.\d+$')

ORDER BY timestamp DESC;

4.2 마스킹 성능 영향 분석

-- 마스킹 처리로 인한 응답 시간 영향 분석
SELECT 
  DATE(timestamp) as log_date,
  service,
  AVG(response_time) as avg_response_time,
  PERCENTILE_CONT(response_time, 0.95) OVER (PARTITION BY service) as p95_response_time,
  COUNT(*) as request_count,
  -- 마스킹 처리가 있는 요청과 없는 요청 비교
  AVG(
    CASE 
      WHEN JSON_EXTRACT_SCALAR(metadata, '$.requestBody') IS NOT NULL 
      THEN response_time 
    END
  ) as avg_response_time_with_body_processing
FROM `dta-cloud-de-dev.log.api_requests`
WHERE timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY)
  AND status_code < 400
GROUP BY log_date, service
ORDER BY log_date DESC, avg_response_time DESC;

5. 데이터 거버넌스 쿼리

5.1 GDPR 준수 확인

-- GDPR 개인정보 처리 기록 확인
SELECT 
  DATE(timestamp) as processing_date,
  service,
  COUNT(DISTINCT user_id) as unique_users_processed,
  COUNT(*) as total_operations,
  -- 개인정보 처리 유형별 분류
  COUNTIF(path LIKE '%/user%') as user_data_operations,
  COUNTIF(path LIKE '%/auth%') as auth_operations,
  COUNTIF(method = 'DELETE') as deletion_operations,
  -- 마스킹되지 않은 개인정보 처리 건수
  COUNTIF(user_id IS NOT NULL AND user_id != '[MASKED]') as unmasked_personal_data
FROM `dta-cloud-de-dev.log.api_requests`
WHERE timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY)
  AND user_id IS NOT NULL
GROUP BY processing_date, service
ORDER BY processing_date DESC;

5.2 데이터 보존 정책 준수 확인

-- 데이터 보존 정책 준수 현황
SELECT 
  table_name,
  TIMESTAMP_DIFF(CURRENT_TIMESTAMP(), max_timestamp, DAY) as days_since_latest,
  TIMESTAMP_DIFF(CURRENT_TIMESTAMP(), min_timestamp, DAY) as retention_days,
  total_rows
FROM (
  SELECT 
    'api_requests' as table_name,
    MAX(timestamp) as max_timestamp,
    MIN(timestamp) as min_timestamp,
    COUNT(*) as total_rows
  FROM `dta-cloud-de-dev.log.api_requests`
  
  UNION ALL
  
  SELECT 
    'system_logs' as table_name,
    MAX(timestamp) as max_timestamp,
    MIN(timestamp) as min_timestamp,
    COUNT(*) as total_rows
  FROM `dta-cloud-de-dev.log.system_logs`
  
  UNION ALL
  
  SELECT 
    'error_logs' as table_name,
    MAX(timestamp) as max_timestamp,
    MIN(timestamp) as min_timestamp,
    COUNT(*) as total_rows
  FROM `dta-cloud-de-dev.log.error_logs`
);

6. BigQuery 스키마 개선 방안

6.1 현재 vs 개선된 스키마 비교

현재 스키마 (문제점)

-- 현재 api_requests 테이블
CREATE TABLE `log.api_requests` (
  timestamp TIMESTAMP,
  user_id STRING,           -- ❌ 원본 UUID 노출
  client_ip STRING,         -- ❌ 원본 IP 주소 노출
  request_headers JSON,     -- ✅ 부분적으로 마스킹됨
  response_body JSON        -- ✅ 부분적으로 마스킹됨
);

개선된 스키마 (권장)

-- 개선된 api_requests 테이블
CREATE TABLE `log.api_requests` (
  timestamp TIMESTAMP,
  -- 사용자 식별 (해시화)
  user_hash STRING,              -- ✅ SHA256 해시값
  user_segment STRING,           -- ✅ 사용자 세그먼트 (premium, basic 등)
  
  -- IP 정보 (메타데이터)
  client_ip_masked STRING,       -- ✅ "192.168.1.***"
  client_region STRING,          -- ✅ "Seoul, KR"
  client_isp STRING,             -- ✅ "KT Corporation"
  client_network_class STRING,   -- ✅ "192.168.0.0/16"
  
  -- 기존 필드
  request_headers JSON,          -- ✅ 마스킹 적용
  response_body JSON,            -- ✅ 마스킹 적용
  method STRING,
  path STRING,
  status_code INTEGER,
  response_time INTEGER
);

6.2 사용자 해시화 기반 추적 쿼리

-- 특정 사용자의 활동 패턴 분석 (해시 기반)
WITH user_hash_lookup AS (
  -- 실시간으로 user_id를 해시화하여 조회
  SELECT SHA256(CONCAT('7194e963-f7eb-43bd-818d', 'YOUR_SALT')) as target_user_hash
)
SELECT 
  DATE(timestamp) as activity_date,
  COUNT(*) as request_count,
  COUNT(DISTINCT path) as unique_endpoints,
  AVG(response_time) as avg_response_time,
  COUNTIF(status_code >= 400) as error_count,
  -- 활동 패턴 분석
  ARRAY_AGG(DISTINCT 
    CONCAT(method, ' ', path) 
    ORDER BY timestamp DESC 
    LIMIT 10
  ) as recent_activities
FROM `dta-cloud-de-dev.log.api_requests` 
CROSS JOIN user_hash_lookup
WHERE user_hash = target_user_hash
  AND timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY)
GROUP BY activity_date
ORDER BY activity_date DESC;

6.3 지역별 보안 분석 쿼리

-- IP 메타데이터 기반 보안 분석
SELECT 
  client_region,
  client_isp,
  COUNT(*) as total_requests,
  COUNT(DISTINCT user_hash) as unique_users,
  -- 보안 이벤트 분석
  COUNTIF(status_code = 401) as auth_failures,
  COUNTIF(status_code = 403) as forbidden_attempts,
  COUNTIF(path LIKE '%admin%') as admin_access_attempts,
  -- 평균 응답 시간 (DDoS 등 탐지)
  AVG(response_time) as avg_response_time,
  STDDEV(response_time) as response_time_variance,
  -- 위험도 계산
  CASE 
    WHEN COUNTIF(status_code = 401) > 100 THEN 'HIGH_RISK'
    WHEN COUNTIF(status_code = 401) > 50 THEN 'MEDIUM_RISK'
    ELSE 'LOW_RISK'
  END as risk_level
FROM `dta-cloud-de-dev.log.api_requests`
WHERE timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 24 HOUR)
  AND client_region IS NOT NULL
GROUP BY client_region, client_isp
HAVING total_requests > 10
ORDER BY auth_failures DESC, total_requests DESC;

6.4 마이그레이션 전략

Phase 1: 기존 데이터 변환

-- 기존 데이터를 새 형식으로 변환
CREATE OR REPLACE VIEW `log.api_requests_enhanced` AS
SELECT 
  timestamp,
  -- 사용자 해시화 (임시 솔루션)
  SHA256(CONCAT(COALESCE(user_id, 'anonymous'), 'YOUR_SALT')) as user_hash,
  -- IP 부분 마스킹
  CONCAT(
    REGEXP_EXTRACT(client_ip, r'^(\d+\.\d+\.\d+)'),
    '.***'
  ) as client_ip_masked,
  -- 지역 정보 추가 (GeoIP 서비스 필요)
  'Unknown' as client_region,  -- TODO: GeoIP 조회로 채우기
  'Unknown' as client_isp,     -- TODO: ISP 정보 조회
  -- 기존 필드
  method,
  path,
  status_code,
  response_time,
  request_headers,
  response_body
FROM `dta-cloud-de-dev.log.api_requests`
WHERE timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY);

Phase 2: 실시간 데이터 파이프라인 수정

// apps/dta-wide-api/src/app/app.module.ts - 로깅 설정 개선
CoreLoggingModule.forRoot({
  httpLogging: {
    // 해시화 및 IP 처리 활성화
    enableUserHashing: true,
    enableIpProcessing: true,
    hashSalt: process.env.USER_HASH_SALT,
    geoIpService: {
      enabled: true,
      provider: 'maxmind' // or 'ip2location'
    }
  }
})

7. 마스킹 개선을 위한 분석 쿼리

7.1 노출된 민감정보 패턴 분석

-- 추가 마스킹이 필요한 필드 식별
WITH field_analysis AS (
  SELECT 
    'metadata.requestBody' as field_path,
    JSON_EXTRACT_SCALAR(metadata, '$.requestBody') as field_value
  FROM `dta-cloud-de-dev.log.api_requests`
  WHERE JSON_EXTRACT_SCALAR(metadata, '$.requestBody') IS NOT NULL
    AND timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 24 HOUR)
  LIMIT 1000
)
SELECT 
  field_path,
  -- 이메일 패턴 탐지
  COUNTIF(REGEXP_CONTAINS(field_value, r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}')) as email_patterns,
  -- 전화번호 패턴 탐지
  COUNTIF(REGEXP_CONTAINS(field_value, r'\d{3}-?\d{3,4}-?\d{4}')) as phone_patterns,
  -- UUID 패턴 탐지
  COUNTIF(REGEXP_CONTAINS(field_value, r'[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}')) as uuid_patterns,
  -- 토큰 패턴 탐지
  COUNTIF(REGEXP_CONTAINS(field_value, r'eyJ[a-zA-Z0-9_-]+\.[a-zA-Z0-9_-]+\.[a-zA-Z0-9_-]+')) as jwt_patterns
FROM field_analysis
GROUP BY field_path;

7. 마스킹 테스트 쿼리

7.1 마스킹 기능 테스트용 쿼리

-- 테스트 환경에서 마스킹 동작 확인
SELECT 
  timestamp,
  request_id,
  path,
  -- 마스킹 전후 비교를 위한 테스트 패턴
  CASE 
    WHEN JSON_EXTRACT_SCALAR(metadata, '$.requestBody.password') = '*****' THEN '✅ 비밀번호 마스킹 성공'
    WHEN JSON_EXTRACT_SCALAR(metadata, '$.requestBody.password') IS NOT NULL THEN '❌ 비밀번호 마스킹 실패'
    ELSE '비밀번호 없음'
  END as password_masking_test,
  CASE 
    WHEN JSON_EXTRACT_SCALAR(request_headers, '$.authorization') = '[REDACTED]' THEN '✅ 인증 헤더 마스킹 성공'
    WHEN JSON_EXTRACT_SCALAR(request_headers, '$.authorization') IS NOT NULL THEN '❌ 인증 헤더 마스킹 실패'
    ELSE '인증 헤더 없음'
  END as auth_header_masking_test
FROM `dta-cloud-de-dev.log.api_requests`
WHERE deploy_env = 'dev'
  AND timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 HOUR)
  AND (
    JSON_EXTRACT_SCALAR(metadata, '$.requestBody.password') IS NOT NULL
    OR JSON_EXTRACT_SCALAR(request_headers, '$.authorization') IS NOT NULL
  )
ORDER BY timestamp DESC
LIMIT 10;

8. 실무 적용 가이드

8.1 단계별 구현 로드맵

Phase 1: 긴급 적용 (1주)

1. 사용자 해시화: user_id → user_hash 변환
2. IP 부분 마스킹: client_ip → client_ip_masked
3. Slack 알림 수정: user_hash 사용

Phase 2: 메타데이터 추가 (2-4주)

1. GeoIP 서비스 연동: 지역/ISP 정보 수집
2. BigQuery 스키마 개선: 새 필드 추가
3. 대시보드 업데이트: 메타데이터 기반 분석

Phase 3: 권한 기반 차등 마스킹 (4-8주)

1. API 응답 마스킹: 권한별 차등 적용
2. 관리자 도구: 해시화된 데이터 조회 인터페이스
3. 모니터링 강화: 자동 알림 시스템

8.2 즉시 확인 필요한 쿼리

1. 민감정보 노출 현황 (섹션 3.1)
2. 사용자 해시화 가능성 테스트 (섹션 6.2)
3. IP 메타데이터 분석 (섹션 6.3)

8.3 정기 모니터링 권장

• 일간: 실시간 위반 사항 모니터링 (섹션 4.1)
• 주간: 마스킹 효과성 분석 (섹션 3.2)
• 월간: GDPR 준수 확인 (섹션 5.1)

8.4 알림 설정 (Google Cloud Monitoring)

-- 알림용 쿼리 (Scheduled Query로 설정)
SELECT 
  COUNT(*) as violations,
  'USER_ID_EXPOSED' as violation_type
FROM `dta-cloud-de-dev.log.api_requests`
WHERE timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 HOUR)
  AND user_id IS NOT NULL 
  AND user_id != '[MASKED]'
  AND LENGTH(user_id) > 10
HAVING violations > 0;

---

관련 문서

• 데이터 마스킹 구현 현황 - 실무적 차별 마스킹 전략 및 구현 방안
• 데이터베이스 보안 구현 가이드 - DB 세션 보안 및 접근 제어
• Platform Requirements - PLT-FR-001 관련

---

변경 이력

버전	날짜	작성자	변경 내용
0.1.0	2025-08-14	bok@weltcorp.com	BigQuery 마스킹된 로그 쿼리 가이드 초기 작성