통신 로그 필드 정규화
60분 안팎
학습 목표
시각·출발지·목적지·상태를 정규화합니다.
개념
로그를 연결하기 전에 형식을 맞추기
앱 접근 기록과 네트워크 관찰 기록을 같은 시간표에 놓으려면 각 필드가 무엇을 뜻하는지 먼저 맞춰야 합니다. 같은 문자열 09:00이라도 시간대가 다르면 다른 순간입니다. 잘못된 주소나 비어 있는 상태를 정상 자료에 섞으면 이후 집계가 왜곡됩니다. 이번 레슨은 합성 HTTP 통신 관찰 배열에서 정상 행을 정규화하고 잘못된 행의 건수를 별도로 출력합니다.
정규화는 읽기 좋은 표로 바꾸는 작업을 넘어 비교 조건을 고정하는 작업입니다. 이 문제는 time, src, dst, port, status라는 다섯 필드만 출력합니다. 추가 입력 필드는 무시하고 본문이나 사용자 정보를 옮기지 않습니다. 필요한 관찰만 남기는 선택으로 로그 분석 과정에서 민감한 값이 퍼지는 범위를 줄입니다. 원래 입력을 지우거나 정상 값으로 꾸며 내지는 않습니다.
필드의 관측 위치 설명하기
src는 해당 관측 위치에서 본 출발지 IP이고 dst는 목적지 IP입니다. 프록시나 NAT를 거친 환경에서는 출발지 주소가 실제 사람의 단말 주소가 아닐 수 있습니다. 따라서 같은 src가 반복된다는 이유만으로 같은 사용자라고 단정하지 않습니다. 미션 보고서에는 관측 위치를 함께 설명하고 웹 사용자 인증 정보와 네트워크 주소를 분리합니다.
port는 이번 계약에서 목적지 포트입니다. 실제 연결 로그에는 출발지 포트도 있을 수 있지만 이 배열에는 없습니다. 443이라고 TLS 검증을 통과했다고 결론 내릴 수는 없습니다. 포트는 목적 서비스의 단서이고 프로토콜과 인증 상태는 별도 근거가 필요합니다. 포트 이름을 service라는 이름으로 바꾸거나 임의로 HTTPS를 채워 넣지 않습니다.
status는 합성 HTTP 응답 코드입니다. 패킷 캡처만으로 HTTPS 응답 코드를 늘 알 수 있다는 뜻이 아닙니다. 이 fixture는 앱 계층 결과까지 포함한 교육용 관찰입니다. DNS rcode, TCP 연결 상태, HTTP status를 동일한 숫자 필드로 합치지 않습니다. 이후 탐지 규칙에서 이 단계들을 구분해야 통신 실패와 자료 조회 실패를 따로 설명할 수 있습니다.
시각을 UTC로 바꾸되 정보는 만들지 않기
time은 초까지 있는 ISO 형태이며 끝에 Z 또는 숫자 오프셋을 요구합니다. 2026-10-09T09:00:00+09:00은 2026-10-09T00:00:00Z와 같은 순간입니다. datetime으로 읽고 timezone.utc로 변환한 뒤 초 단위 문자열을 출력합니다. 날짜가 넘어가는 변환도 라이브러리에 맡기며 시간에서 9를 단순히 빼는 방식으로 계산하지 않습니다.
시간대가 없는 2026-10-09T09:00:00은 INVALID 행으로 셉니다. 실행한 컴퓨터의 지역 시간대를 추측해서 덧붙이지 않습니다. 원 수집기의 시간대를 확인할 근거가 없기 때문입니다. 2월 30일 같은 존재하지 않는 날짜도 파싱에 실패합니다. 이런 행을 현재 시간으로 대체하면 사건 순서가 바뀌므로 분석용 기록으로 받아들이지 않습니다.
이번 계약은 소수 초와 윤초를 다루지 않으며 숫자 오프셋의 시간은 23 이하, 분은 59 이하입니다. 실무 입력 형식이 넓으면 지원 범위를 별도로 확장해야 합니다. 정규식은 형태를 검사하고 datetime은 실제 날짜를 검사합니다. 정규식에 숫자 여덟 자리가 들어 있다는 이유만으로 날짜가 유효한 것은 아니라는 차이를 확인합니다.
숫자 범위와 형식을 함께 검사하기
port는 정수 1~65535, status는 정수 100~599입니다. 0 포트와 600 상태는 이 관찰 계약 밖입니다. status가 문자열 "200"이면 숫자로 자동 변환하지 않고 잘못된 행으로 셉니다. 수집자가 계약을 어긴 사실을 가리지 않기 위한 선택입니다. 필요하다면 원 수집기에서 변환 규칙을 정의하지만 이 레슨의 정규화에서 조용히 보정하지 않습니다.
Python에서 bool은 int와 관계가 있으므로 isinstance(True, int)는 True입니다. 포트 True를 1로 받아들이지 않으려면 type(value) is int를 사용합니다. 타입 검사와 범위 검사는 둘 다 필요합니다. 65535는 허용 끝값이고 65536은 제외됩니다. 응답 코드도 100과 599를 받아들이지만 실제 각 숫자의 표준 등록 의미를 이 함수가 판별하는 것은 아닙니다.
주소는 문자열 형태의 IP만 받습니다. IPv4와 IPv6를 ipaddress로 읽고 str로 다시 쓰면 IPv6의 불필요한 0이 정리됩니다. 예를 들어 2001:0db8::1은 2001:db8::1이 됩니다. 도메인 이름을 여기서 DNS로 바꾸지 않습니다. 같은 행을 분석할 때마다 외부 조회 결과가 달라지지 않도록 입력의 실제 관측값만 다룹니다.
유효하지 않은 행을 분리하는 반환 계약
normalize는 유효한 행이면 다섯 필드 객체를 반환하고 아니면 None을 반환합니다. 바깥 반복문은 None을 invalid 건수에 더합니다. 정상 행은 입력 순서대로 records에 추가합니다. 하나의 행에 잘못된 필드가 여러 개 있어도 invalid는 한 번만 증가합니다. 이것은 잘못된 행 수이지 오류 필드 개수라는 의미가 아닙니다.
결측 키가 있거나 행 자체가 객체가 아니어도 해당 행만 제외합니다. 정상 404 행은 실패한 업무 결과지만 로그 형식은 올바르므로 records에 남습니다. 반면 200처럼 보이는 문자열 상태는 잘못된 형식입니다. 업무 성공과 로그 유효성을 분리하면 실패 요청을 모두 버려 보안 경보의 근거를 잃는 실수를 피할 수 있습니다.
중복 행을 자동 제거하거나 시각 순으로 정렬하지 않습니다. 같은 요청의 반복인지 실제 별도 요청인지 결정할 식별자가 아직 없기 때문입니다. 두 개의 동일 행은 records에 두 번 남습니다. 자료의 형식을 맞추는 일과 사건을 합치는 일은 서로 다른 단계입니다. 다음 레슨에서 요청 식별자를 연결할 때도 단순한 시각 일치만으로 같은 사건이라고 합치지 않습니다.
starter와 오류 위치 확인하기
starter는 입력 배열을 읽고 건수를 출력하는 구조를 제공하지만 normalize가 None만 반환합니다. 따라서 빈 배열 검사는 맞을 수 있어도 정상 한 행 검사는 실패합니다. 필수 키, 타입과 범위, 시간대 형태, 실제 날짜와 주소 검사를 구현합니다. try 범위는 변환 부분으로 제한하고 ValueError·TypeError만 행 제외로 다룹니다. 예상하지 못한 코드 오류를 전체 except로 숨기지 않습니다.
records가 비어 있고 invalid가 전체 건수와 같으면 정상 반환 경로가 빠졌는지 확인합니다. 같은 순간의 두 시각이 다르게 나오면 astimezone 적용 여부를 봅니다. NameError는 잘못된 변수 이름이므로 오류를 None으로 바꾸기보다 코드를 고칩니다. 테스트의 기대 JSON을 지우면 잘못된 필터가 통과하므로 입력 하나를 손으로 계산해 비교합니다.
자료 품질을 보고서에 드러내기
출력은 records와 invalid를 가진 JSON 객체 한 개입니다. 빈 배열은 records 빈 목록, invalid 0입니다. 보고서에는 수집한 전체 행 수와 받아들인 행 수, 제외한 행 수를 함께 적습니다. 제외 사유별 통계는 이번 계약에 없으므로 “대부분 시간 오류였다”처럼 숫자 근거 없는 문장은 쓰지 않습니다. 필요한 경우 후속 과제로 사유 코드를 추가한다고 설명합니다.
이번 결과는 네트워크 전체를 감시하거나 공격 여부를 판정하지 않습니다. 시각 정규화와 필드 검증이 완료된 합성 자료입니다. 404 증가를 탐지 규칙으로 쓰려면 정상 사용자 행동, 관측 누락, 기간과 요청 식별자 같은 추가 맥락이 필요합니다. 파일 입출력과 예외 처리의 문법은 더 읽기로 보내고 여기서는 계약·경계·제외 수치를 근거로 재현 가능한 입력을 만드는 데 집중합니다.
따라하기
UTC 변환 확인
합성 시각을 실제 datetime 계산으로 바꿉니다.
from datetime import datetime, timezone
for stamp in ['2026-10-09T09:00:00+09:00','2026-10-08T20:00:00-04:00']:
print(datetime.fromisoformat(stamp).astimezone(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ'))실행 결과
2026-10-09T00:00:00Z 2026-10-09T00:00:00Z
불리언과 정수 분리
포트 범위와 정확한 타입 조건을 함께 비교합니다.
for port in [1,65535,65536,True,'443']:
print(repr(port), type(port) is int and 1 <= port <= 65535)실행 결과
1 True 65535 True 65536 False True False '443' False
정상과 제외 행을 함께 분석
정상 404는 유지하고 포트 0은 제외합니다.
import sys, json, ipaddress, re
from datetime import datetime, timezone
def normalize(row):
if not isinstance(row, dict) or not all(k in row for k in ("time", "src", "dst", "port", "status")):
return None
if type(row["port"]) is not int or not 1 <= row["port"] <= 65535:
return None
if type(row["status"]) is not int or not 100 <= row["status"] <= 599:
return None
if not isinstance(row["time"], str) or not re.fullmatch(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(Z|[+-]\d{2}:\d{2})", row["time"]):
return None
try:
if not isinstance(row["src"], str) or not isinstance(row["dst"], str):
return None
src, dst = str(ipaddress.ip_address(row["src"])), str(ipaddress.ip_address(row["dst"]))
dt = datetime.fromisoformat(row["time"].replace("Z", "+00:00"))
# The contract excludes offset values beyond 23:59.
if not row["time"].endswith("Z") and (int(row["time"][-5:-3]) > 23 or int(row["time"][-2:]) > 59):
return None
stamp = dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
return {"time": stamp, "src": src, "dst": dst, "port": row["port"], "status": row["status"]}
except (ValueError, TypeError):
return None
rows = [{'time': '2026-10-09T09:00:00+09:00', 'src': '192.0.2.1', 'dst': '192.0.2.10', 'port': 443, 'status': 200}, {'time': '2026-10-09T09:00:00+09:00', 'src': '192.0.2.1', 'dst': '192.0.2.10', 'port': 0, 'status': 200}, {'time': '2026-10-09T09:00:00+09:00', 'src': '192.0.2.1', 'dst': '192.0.2.10', 'port': 443, 'status': 404}]
valid = []
invalid = 0
for row in rows:
item = normalize(row)
if item is None:
invalid += 1
else:
valid.append(item)
print(json.dumps({"records": valid, "invalid": invalid}, ensure_ascii=False))
실행 결과
{"records": [{"time": "2026-10-09T00:00:00Z", "src": "192.0.2.1", "dst": "192.0.2.10", "port": 443, "status": 200}, {"time": "2026-10-09T00:00:00Z", "src": "192.0.2.1", "dst": "192.0.2.10", "port": 443, "status": 404}], "invalid": 1}
확인 문제
실습
JSON 배열을 읽어 records와 invalid 객체를 출력합니다. time은 초 단위 Z 또는 오프셋이 있는 시각을 UTC로, src·dst는 IP 정규 문자열로 바꿉니다. port는 정수 1~65535, status는 정수 100~599이고 bool·숫자 문자열은 제외합니다. 결측·형식 오류 행은 한 번만 invalid에 셉니다. 정상 404와 중복 행은 유지하고 입력 순서대로 출력합니다. 추가 필드는 출력하지 않습니다. JSON 문법과 전체 배열 형식은 올바르게 제공됩니다.
모범 답안
import sys, json, ipaddress, re
from datetime import datetime, timezone
def normalize(row):
if not isinstance(row, dict) or not all(k in row for k in ("time", "src", "dst", "port", "status")):
return None
if type(row["port"]) is not int or not 1 <= row["port"] <= 65535:
return None
if type(row["status"]) is not int or not 100 <= row["status"] <= 599:
return None
if not isinstance(row["time"], str) or not re.fullmatch(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(Z|[+-]\d{2}:\d{2})", row["time"]):
return None
try:
if not isinstance(row["src"], str) or not isinstance(row["dst"], str):
return None
src, dst = str(ipaddress.ip_address(row["src"])), str(ipaddress.ip_address(row["dst"]))
dt = datetime.fromisoformat(row["time"].replace("Z", "+00:00"))
# The contract excludes offset values beyond 23:59.
if not row["time"].endswith("Z") and (int(row["time"][-5:-3]) > 23 or int(row["time"][-2:]) > 59):
return None
stamp = dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
return {"time": stamp, "src": src, "dst": dst, "port": row["port"], "status": row["status"]}
except (ValueError, TypeError):
return None
rows = json.load(sys.stdin)
valid = []
invalid = 0
for row in rows:
item = normalize(row)
if item is None:
invalid += 1
else:
valid.append(item)
print(json.dumps({"records": valid, "invalid": invalid}, ensure_ascii=False))
더 읽기
면접 질문
- 시간대 없는 로그와 정상 404 로그를 각각 어떻게 처리합니까?