diff --git a/app/common/validation.py b/app/common/validation.py new file mode 100644 index 0000000..743c627 --- /dev/null +++ b/app/common/validation.py @@ -0,0 +1,49 @@ +"""Shared wire schema. Semantic storage validation belongs to the worker.""" +import json +import math + +STRING_FIELDS = ("raw_id", "device_id", "caller_num", "svc", "lac", "ci") +INTEGER_FIELDS = ("rat", "epoch_start", "epoch_end", "up_bytes", "down_bytes") +REQUIRED_FIELDS = STRING_FIELDS + INTEGER_FIELDS + + +class PayloadError(ValueError): + def __init__(self, reason, detail, field=None): + super().__init__(detail) + self.reason = reason + self.field = field + + +def parse_payload(data): + def reject_constant(value): + raise ValueError(f"Non-finite number: {value}") + def finite_float(value): + number = float(value) + if not math.isfinite(number): + raise ValueError(f"Non-finite number: {value}") + return number + try: + payload = json.loads(data.decode("utf-8"), parse_constant=reject_constant, parse_float=finite_float) + except (UnicodeError, ValueError, RecursionError) as exc: + raise PayloadError("invalid_json", str(exc)) from exc + if not isinstance(payload, dict): + raise PayloadError("invalid_payload", "Expected a JSON object") + return payload + + +def validate_structure(payload): + for field in REQUIRED_FIELDS: + if payload.get(field) is None: + raise PayloadError("missing_required_field", f"Field {field} is required", field) + for field in STRING_FIELDS: + if not isinstance(payload[field], str) or not payload[field].strip(): + raise PayloadError("invalid_payload", f"Field {field} must be a nonempty string", field) + try: + payload[field].encode("utf-8") + except UnicodeError as exc: + raise PayloadError("invalid_payload", f"Field {field} contains invalid Unicode", field) from exc + for field in INTEGER_FIELDS: + if type(payload[field]) is not int: + raise PayloadError("invalid_payload", f"Field {field} must be an integer", field) + if payload.get("src_ip") is not None and not isinstance(payload["src_ip"], str): + raise PayloadError("invalid_payload", "Field src_ip must be a string or null", "src_ip") diff --git a/app/worker/normalizer.py b/app/worker/normalizer.py index f6842dc..c4d6f2b 100644 --- a/app/worker/normalizer.py +++ b/app/worker/normalizer.py @@ -1,4 +1,6 @@ import json +from ipaddress import IPv4Address +from app.common.validation import PayloadError, parse_payload, validate_structure import uuid from dataclasses import dataclass from datetime import datetime, timezone @@ -58,9 +60,9 @@ class NormalizedRecord: self.bytes_downloaded, self.total_bytes, self.duration_seconds, - datetime.strptime(self.session_start_time, "%Y-%m-%d %H:%M:%S"), - datetime.strptime(self.session_end_time, "%Y-%m-%d %H:%M:%S"), - self.source_ip, + datetime.strptime(self.session_start_time, "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc), + datetime.strptime(self.session_end_time, "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc), + IPv4Address(self.source_ip), ) @@ -105,96 +107,51 @@ def normalize_cdr_payload( (record, failure_reason, failed_field, raw_id) If record is None, failure_reason describes why it should be routed to DLQ. """ - # 1. Parse JSON try: - if isinstance(raw_payload_bytes, bytes): - payload_str = raw_payload_bytes.decode("utf-8") - else: - payload_str = str(raw_payload_bytes) - payload: Dict[str, Any] = json.loads(payload_str) - except Exception: + payload = parse_payload(raw_payload_bytes) + except PayloadError: return None, "malformed_json", None, None - - if not isinstance(payload, dict): - return None, "malformed_json", None, None - raw_id = payload.get("raw_id") - raw_id_str = str(raw_id) if raw_id is not None else None - - # 2. Check required fields - for field in REQUIRED_FIELDS: - if field not in payload or payload[field] is None: - return None, "missing_required_field", field, raw_id_str - - # 3. Check timestamps (epoch_start <= epoch_end) + raw_id_str = raw_id if isinstance(raw_id, str) else None try: - epoch_start = int(payload["epoch_start"]) - epoch_end = int(payload["epoch_end"]) - if epoch_start > epoch_end: - return None, "invalid_epoch", "epoch_start", raw_id_str - except (ValueError, TypeError): + validate_structure(payload) + except PayloadError as exc: + reason = "missing_required_field" if exc.reason == "missing_required_field" else ( + "invalid_epoch" if exc.field in ("epoch_start", "epoch_end") else + "invalid_rat_value" if exc.field == "rat" else "normalization_error") + return None, reason, exc.field, raw_id_str + for field in ("epoch_start", "epoch_end"): + if not 0 <= payload[field] <= 2**32 - 1: + return None, "invalid_epoch", field, raw_id_str + epoch_start, epoch_end = payload["epoch_start"], payload["epoch_end"] + if epoch_start > epoch_end: return None, "invalid_epoch", "epoch_start", raw_id_str - - # 4. Check bytes (must be >= 0) - try: - up_bytes = int(payload["up_bytes"]) - down_bytes = int(payload["down_bytes"]) - if up_bytes < 0: - return None, "normalization_error", "up_bytes", raw_id_str - if down_bytes < 0: - return None, "normalization_error", "down_bytes", raw_id_str - except (ValueError, TypeError): - return None, "normalization_error", "bytes", raw_id_str - - # 5. Radio Access Type mapping - rat_raw = payload.get("rat") - try: - rat_int = int(rat_raw) - if rat_int not in RAT_MAP: - return None, "invalid_rat_value", "rat", raw_id_str - radio_access_type = RAT_MAP[rat_int] - except (ValueError, TypeError): + for field in ("up_bytes", "down_bytes"): + if not 0 <= payload[field] <= 2**64 - 1: + return None, "normalization_error", field, raw_id_str + up_bytes, down_bytes = payload["up_bytes"], payload["down_bytes"] + if up_bytes + down_bytes > 2**64 - 1: + return None, "normalization_error", "total_bytes", raw_id_str + if payload["rat"] not in RAT_MAP: return None, "invalid_rat_value", "rat", raw_id_str - - # 6. E.164 phone normalization - caller_num = str(payload["caller_num"]) - msisdn = normalize_phone_number(caller_num) - if not msisdn: + msisdn = normalize_phone_number(payload["caller_num"]) + if msisdn is None: return None, "normalization_error", "caller_num", raw_id_str - - # 7. Field mappings and derivations - imsi = str(payload["device_id"]) - service_type = str(payload["svc"]) - cell_id = f"{payload['lac']}_{payload['ci']}" - total_bytes = up_bytes + down_bytes - duration_seconds = epoch_end - epoch_start - - start_dt = datetime.fromtimestamp(epoch_start, tz=timezone.utc).strftime( - "%Y-%m-%d %H:%M:%S" - ) - end_dt = datetime.fromtimestamp(epoch_end, tz=timezone.utc).strftime( - "%Y-%m-%d %H:%M:%S" - ) - - src_ip = str(payload.get("src_ip") or "0.0.0.0") - - # Deterministic UUIDv5 cdr_id from raw_id - cdr_id = str(uuid.uuid5(uuid.NAMESPACE_DNS, raw_id_str)) - + try: + source_ip = str(IPv4Address(payload["src_ip"] if payload.get("src_ip") is not None else "0.0.0.0")) + except (ValueError, TypeError): + return None, "normalization_error", "src_ip", raw_id_str + try: + start_dt = datetime.fromtimestamp(epoch_start, timezone.utc).strftime("%Y-%m-%d %H:%M:%S") + end_dt = datetime.fromtimestamp(epoch_end, timezone.utc).strftime("%Y-%m-%d %H:%M:%S") + except (ValueError, OverflowError, OSError): + return None, "invalid_epoch", "epoch_start", raw_id_str record = NormalizedRecord( - cdr_id=cdr_id, - raw_id=raw_id_str, - msisdn=msisdn, - imsi=imsi, - service_type=service_type, - radio_access_type=radio_access_type, - cell_id=cell_id, - bytes_uploaded=up_bytes, - bytes_downloaded=down_bytes, - total_bytes=total_bytes, - duration_seconds=duration_seconds, - session_start_time=start_dt, - session_end_time=end_dt, - source_ip=src_ip, + cdr_id=str(uuid.uuid5(uuid.NAMESPACE_DNS, raw_id)), raw_id=raw_id, + msisdn=msisdn, imsi=payload["device_id"], service_type=payload["svc"], + radio_access_type=RAT_MAP[payload["rat"]], cell_id=f"{payload['lac']}_{payload['ci']}", + bytes_uploaded=up_bytes, bytes_downloaded=down_bytes, total_bytes=up_bytes+down_bytes, + duration_seconds=epoch_end-epoch_start, session_start_time=start_dt, + session_end_time=end_dt, source_ip=source_ip, ) - return record, None, None, raw_id_str + return record, None, None, raw_id