| """Strict input validation for the RMTI API. |
| |
| Design rules: |
| * Every request body is checked against an explicit allowlist of fields. |
| Unknown fields are rejected (HTTP 400), not silently ignored. This also |
| acts as a guard against someone accidentally posting identifiers |
| (names, record numbers, dates of birth) to the oncology endpoints. |
| * Every numeric input is range-checked BEFORE it reaches the framework |
| code, because core.rmti_core.score() itself does not validate ranges. |
| * Booleans, NaN, Infinity and strings-that-look-like-numbers are rejected. |
| """ |
| import math |
|
|
| MAX_ITEMS = 100 |
| MAX_HAZARDS = 10 |
| MAX_LABEL_LEN = 60 |
| N_HARD_MAX = 10_000_000_000 |
|
|
|
|
| class ApiError(Exception): |
| def __init__(self, status, code, message): |
| super().__init__(message) |
| self.status, self.code, self.message = status, code, message |
|
|
|
|
| def bad(code, message): |
| return ApiError(400, code, message) |
|
|
|
|
| def require_object(body, what="request body"): |
| if not isinstance(body, dict): |
| raise bad("invalid_body", f"{what} must be a JSON object") |
| return body |
|
|
|
|
| def check_fields(d, allowed, where="request"): |
| unknown = sorted(set(d) - set(allowed)) |
| if unknown: |
| raise bad("unknown_field", |
| f"Unexpected field(s) in {where}: {', '.join(map(str, unknown))}. " |
| f"Allowed: {', '.join(sorted(allowed))}. " |
| "Do not submit names, record numbers or other identifiers.") |
|
|
|
|
| def num(d, key, lo, hi, *, required=True, default=None, where="request"): |
| if key not in d: |
| if required: |
| raise bad("missing_field", f"'{key}' is required in {where}") |
| return default |
| v = d[key] |
| if isinstance(v, bool) or not isinstance(v, (int, float)): |
| raise bad("invalid_type", f"'{key}' must be a number in {where}") |
| if not math.isfinite(v): |
| raise bad("invalid_value", f"'{key}' must be finite in {where}") |
| if v < lo or v > hi: |
| raise bad("out_of_range", f"'{key}'={v} is outside [{lo}, {hi}] in {where}") |
| return v |
|
|
|
|
| def integer_in(d, key, choices, *, required=True, default=None, where="request"): |
| if key not in d: |
| if required: |
| raise bad("missing_field", f"'{key}' is required in {where}") |
| return default |
| v = d[key] |
| if isinstance(v, bool) or not isinstance(v, (int, float)) or v != int(v): |
| raise bad("invalid_type", f"'{key}' must be an integer in {where}") |
| if int(v) not in choices: |
| raise bad("out_of_range", f"'{key}'={v} must be one of {sorted(choices)} in {where}") |
| return int(v) |
|
|
|
|
| def label(d, where="item"): |
| v = d.get("label") |
| if not isinstance(v, str) or not (1 <= len(v) <= MAX_LABEL_LEN): |
| raise bad("invalid_label", f"'label' must be a string of 1-{MAX_LABEL_LEN} characters in {where}") |
| if any(ord(c) < 32 for c in v): |
| raise bad("invalid_label", f"'label' must not contain control characters in {where}") |
| return v |
|
|
|
|
| def rho_pair(rho_now, rho_after, where): |
| """rho_now == 1 would divide by zero in the risk-reduction formula.""" |
| if rho_after is not None and rho_now >= 1: |
| raise bad("invalid_value", |
| f"'rho_now' must be < 1 when 'rho_after' is given (risk reduction is undefined) in {where}") |
|
|
|
|
| def items_list(body): |
| require_object(body) |
| check_fields(body, {"items"}, "request body") |
| items = body.get("items") |
| if not isinstance(items, list) or not items: |
| raise bad("invalid_body", "'items' must be a non-empty array") |
| if len(items) > MAX_ITEMS: |
| raise bad("too_many_items", f"'items' may contain at most {MAX_ITEMS} entries") |
| return items |
|
|
