okta_compat.py

# The Okta software accompanied by this notice is provided pursuant to the following terms:
# Copyright © 2026-Present, Okta, Inc.
# Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License.
# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0.
# Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and limitations under the License.

r"""Runtime compatibility patches for over-strict ``okta`` SDK models.

Written against **okta==3.4.4** (released 2026-07-01 — the latest version, and the
one this project pins), whose ~1786 model classes are Pydantic v2 ``BaseModel``\s
auto-generated by OpenAPI Generator from Okta's **OpenAPI 5.1.0** description.

Several of those generated models are *stricter than the API they describe*, so
legitimate Okta responses raise ``pydantic.ValidationError`` deep inside the SDK —
in ``ApiClient.__deserialize_model`` → ``klass.from_dict(data)`` — before any MCP
tool code runs, aborting the whole request.  The models live inside a pinned
third-party dependency and cannot be edited in-tree, so we relax them at runtime.

There is no newer ``okta-sdk-python`` release to move to: v3.4.4 is the latest, and
defects B, C and D have no released fix at all.

Why targeted patches and not a blanket relaxation
-------------------------------------------------
Only the fields listed below are touched.  Most generated models are *request*
models where local required-field validation is genuinely useful — relaxing them
wholesale would turn a clear local error into an opaque API 400, and setting
``extra="allow"`` would discard the validation signal entirely.  The general
safety net for unknown future spec drift is per-item tolerance on **list**
responses, implemented in :mod:`okta_mcp_server.utils.tolerant_deserialization`.

Pydantic v2 mechanics
---------------------
Two things must both happen for an inherited field:

* Each subclass gets its **own** ``model_fields`` dict holding its **own**
  ``FieldInfo`` object, even for fields it does not redeclare.  Patching only a
  base class therefore does **not** fix subclasses that already exist — hence the
  subclass sweep in :func:`relax_field_everywhere`.
* Patching ``cls.__annotations__`` *does* reach subclasses created **after** the
  patch, because Pydantic collects inherited fields by walking the MRO's
  ``__annotations__``.

``__subclasses__()`` only sees classes that have already been imported and
``okta.models`` imports lazily, so :func:`_patch_defect_b` force-imports the
concrete ``Policy`` subclasses rather than trusting import order.

Upstream tracking — each patch is removable once the matching fix ships:

* **Defect A** — ``SamlApplicationSettingsSignOn`` required ``StrictBool``\s:
  okta/okta-sdk-python#546 (open; reports this still broken on 3.4.4).
  okta/okta-sdk-python#536 (closed) was partially fixed by merged PR #542 in
  v3.4.3, which removed 10 of 16 required constraints but left these 5.
  MCP-side symptom: okta/okta-mcp-server#48 (open).
* **Defect B** — ``Policy.embedded`` inner value type: okta/okta-mcp-server#100
  (open), fix pending in okta/okta-mcp-server PR #102 (open).  No
  okta-sdk-python issue exists yet.
* **Defect C** — closed authenticator enum rejects ``smart_card_idp``:
  okta/okta-mcp-server#101 (open), fix pending in PR #102 (open).  No
  okta-sdk-python issue exists yet.
* **Defect D** — ``UserTypeCondition`` required lists: **unreported upstream as
  of 2026-08-04**, in any repo.
* **LogSecurityContext.user_behaviors** — pre-existing patch, migrated here from
  ``tools/system_logs/system_logs.py`` so all SDK compat lives in one place.
  Behavior is unchanged.
"""

from __future__ import annotations

import importlib
import typing as t

from loguru import logger

__all__ = [
    "apply_okta_model_compat",
    "relax_field",
    "relax_field_everywhere",
]


# ---------------------------------------------------------------------------
# Generic relaxation helpers
# ---------------------------------------------------------------------------

def relax_field(model_cls: type, field_name: str, annotation: t.Any, default: t.Any = None) -> bool:
    """Widen one Pydantic v2 field on ``model_cls`` and rebuild its schema.

    Updates three things that must agree:

    1. ``model_cls.__annotations__[field_name]`` — what Pydantic reads when it
       collects fields for subclasses created *later*.
    2. ``model_cls.model_fields[field_name].annotation`` — the live schema source
       for this class.
    3. ``.default`` (and ``.default_factory``) — in Pydantic v2 a field is
       required precisely while its default and default factory are both unset.
       Assigning ``default=None`` is what makes a previously-required field
       optional; changing only ``.annotation`` leaves it required and merely
       swaps a ``bool_type`` error for a ``missing`` one.

    ``model_rebuild(force=True)`` then regenerates the core schema.

    Args:
        model_cls: The generated SDK model class to patch.
        field_name: The *Python* field name (snake_case), not the JSON alias.
        annotation: The replacement annotation, e.g. ``Optional[bool]``.
        default: The replacement default.  ``None`` makes the field optional.

    Returns:
        ``True`` if the field existed and was patched; ``False`` if it is absent,
        e.g. because a future SDK version dropped or renamed it.
    """
    model_fields = getattr(model_cls, "model_fields", None)
    if not model_fields or field_name not in model_fields:
        return False

    # Reading cls.__annotations__ on a class that declares none creates an empty
    # dict on that class (Python >= 3.10), so this never leaks into a base class.
    model_cls.__annotations__[field_name] = annotation

    field_info = model_fields[field_name]
    field_info.annotation = annotation
    field_info.default = default
    field_info.default_factory = None

    model_cls.model_rebuild(force=True)
    return True


def _iter_subclasses(cls: type) -> t.Iterator[type]:
    """Yield every already-created subclass of ``cls``, depth-first."""
    for sub in cls.__subclasses__():
        yield sub
        yield from _iter_subclasses(sub)


def relax_field_everywhere(
    model_cls: type,
    field_name: str,
    annotation: t.Any,
    default: t.Any = None,
    extra_classes: t.Sequence[type] = (),
) -> list[str]:
    """Apply :func:`relax_field` to ``model_cls`` and every subclass carrying the field.

    Required whenever the over-strict field is inherited — see the Pydantic v2
    section of the module docstring.  ``extra_classes`` lets a caller name
    concrete subclasses explicitly, so the patch does not silently depend on
    which SDK modules happened to be imported first.

    Returns:
        Names of the classes actually patched, for debug logging.
    """
    targets: list[type] = [model_cls, *extra_classes, *_iter_subclasses(model_cls)]
    patched: list[str] = []
    seen: set[int] = set()
    for cls in targets:
        if id(cls) in seen:
            continue
        seen.add(id(cls))
        if relax_field(cls, field_name, annotation, default):
            patched.append(cls.__name__)
    return patched


# ---------------------------------------------------------------------------
# Individual patches
# ---------------------------------------------------------------------------

#: Concrete ``Policy`` subclasses shipped by okta==3.4.4.  Importing a module is
#: enough to create its class and make it visible to ``Policy.__subclasses__()``.
_POLICY_SUBCLASS_MODULES = (
    "access_policy",
    "authenticator_enrollment_policy",
    "device_signal_collection_policy",
    "entity_risk_policy",
    "idp_discovery_policy",
    "okta_sign_on_policy",
    "password_policy",
    "post_auth_session_policy",
    "profile_enrollment_policy",
)


def _patch_defect_a() -> str:
    """Defect A — ``SamlApplicationSettingsSignOn``'s 5 required ``StrictBool`` fields.

    They are the *only* required fields on the class (the other ~25 already
    default to ``None``), yet Okta routinely omits them from SAML app responses.
    Both entry points must be covered: ``model_validate`` on a payload missing
    the keys reports ``Field required``, while the SDK's own ``from_dict`` reads
    ``obj.get(...)`` and so reports ``Input should be a valid boolean`` for
    ``None``.  Relaxing annotation *and* default handles both.

    Upstream: okta/okta-sdk-python#546 (open), okta/okta-mcp-server#48 (open).
    Removable once the SDK drops the required constraint on these five.
    """
    from okta.models.saml_application_settings_sign_on import SamlApplicationSettingsSignOn

    names = (
        "allow_multiple_acs_endpoints",
        "assertion_signed",
        "honor_force_authn",
        "request_compressed",
        "response_signed",
    )
    relaxed = [n for n in names if relax_field(SamlApplicationSettingsSignOn, n, t.Optional[bool])]
    return f"relaxed {len(relaxed)}/{len(names)} sign-on booleans to Optional[bool]: {relaxed}"


def _patch_defect_b() -> str:
    """Defect B — ``_embedded`` declared as ``Dict[str, Dict[str, Any]]``.

    Okta returns scalars inside the HAL envelope, e.g.
    ``{"_embedded": {"resourceType": "APP"}}``.  ``_embedded`` is transport
    metadata rather than business data, so widening its value type costs no
    meaningful validation.  Applied to every SDK model that declares this same
    narrow shape (located with ``grep -rn 'alias="_embedded"'``), each with its
    own subclass sweep.

    ``AccessPolicy`` is the class observed failing against a real tenant; it is
    covered twice over — force-imported *and* named in ``extra_classes`` — because
    ``__subclasses__()`` alone is import-order dependent.

    Upstream: okta/okta-mcp-server#100 (open), PR #102 (open).
    Removable once the SDK regenerates ``_embedded`` as ``Dict[str, Any]``.
    """
    # Importing each module creates its class and registers it as a Policy
    # subclass, so the sweep below finds it regardless of import order.
    for mod_name in _POLICY_SUBCLASS_MODULES:
        try:
            importlib.import_module(f"okta.models.{mod_name}")
        except ImportError:  # pragma: no cover — a future SDK dropped the model
            logger.debug(f"[okta_compat] okta.models.{mod_name} not present; skipping")

    from okta.models.access_policy import AccessPolicy
    from okta.models.create_or_update_policy import CreateOrUpdatePolicy
    from okta.models.policy import Policy
    from okta.models.policy_common import PolicyCommon
    from okta.models.role import Role
    from okta.models.user_factor import UserFactor
    from okta.models.user_factor_yubikey_otp_token import UserFactorYubikeyOtpToken

    patched_type = t.Optional[t.Dict[str, t.Any]]
    patched: list[str] = relax_field_everywhere(
        Policy, "embedded", patched_type, extra_classes=(AccessPolicy,)
    )
    for base in (PolicyCommon, CreateOrUpdatePolicy, UserFactor, UserFactorYubikeyOtpToken, Role):
        patched += relax_field_everywhere(base, "embedded", patched_type)

    return f"relaxed _embedded to Optional[Dict[str, Any]] on {len(set(patched))} classes: {sorted(set(patched))}"


def _patch_defect_c() -> str:
    """Defect C — closed 15-member authenticator enum rejects ``smart_card_idp``.

    ``AuthenticatorEnrollmentPolicyAuthenticatorType`` omits ``smart_card_idp``
    (PIV/CAC).  Rather than hand-adding the one missing member — which would break
    again on the next authenticator type Okta ships — the ``key`` field is relaxed
    to a plain string, so any key the API returns is accepted instead of us
    maintaining a second, hand-kept enum that drifts from the API.  This matches
    the fix in okta/okta-mcp-server PR #102.

    Upstream: okta/okta-mcp-server#101 (open), PR #102 (open).
    Removable once the SDK regenerates the enum as open-ended.
    """
    from okta.models.authenticator_enrollment_policy_authenticator_settings import (
        AuthenticatorEnrollmentPolicyAuthenticatorSettings,
    )

    ok = relax_field(AuthenticatorEnrollmentPolicyAuthenticatorSettings, "key", t.Optional[str])
    return f"relaxed authenticator settings key to Optional[str]: {ok}"


def _patch_defect_d() -> str:
    """Defect D — policy condition models requiring lists that Okta sends as ``null``.

    Okta returns ``{"userType": {"exclude": null, "include": null}}``, and *both*
    fields are declared as required lists, so both fail.  Two sibling condition
    models are over-strict the same way.  Siblings that are already correctly
    Optional are deliberately left alone: ``group_condition``, ``user_condition``,
    ``policy_people_condition``, ``policy_network_condition`` and
    ``platform_policy_rule_condition``.

    Upstream: unreported as of 2026-08-04, in any repo.
    Removable once the SDK marks these condition members nullable.
    """
    from okta.models.detected_risk_events import DetectedRiskEvents
    from okta.models.risk_detection_types_policy_rule_condition import (
        RiskDetectionTypesPolicyRuleCondition,
    )
    from okta.models.user_identifier_condition_evaluator_pattern import (
        UserIdentifierConditionEvaluatorPattern,
    )
    from okta.models.user_identifier_policy_rule_condition import (
        UserIdentifierPolicyRuleCondition,
    )
    from okta.models.user_type_condition import UserTypeCondition

    targets: tuple[tuple[type, str, t.Any], ...] = (
        (UserTypeCondition, "exclude", t.Optional[t.List[str]]),
        (UserTypeCondition, "include", t.Optional[t.List[str]]),
        (RiskDetectionTypesPolicyRuleCondition, "exclude", t.Optional[t.List[DetectedRiskEvents]]),
        (RiskDetectionTypesPolicyRuleCondition, "include", t.Optional[t.List[DetectedRiskEvents]]),
        (
            UserIdentifierPolicyRuleCondition,
            "patterns",
            t.Optional[t.List[UserIdentifierConditionEvaluatorPattern]],
        ),
    )
    relaxed = [
        f"{cls.__name__}.{name}"
        for cls, name, annotation in targets
        if relax_field(cls, name, annotation)
    ]
    return f"relaxed {len(relaxed)}/{len(targets)} condition fields: {relaxed}"


def _patch_log_security_context() -> str:
    """``LogSecurityContext.user_behaviors`` — migrated from ``system_logs.py``.

    When Behavior Detection is enabled the Okta API returns ``userBehaviors`` as
    ``List[dict]``, but the model declares ``List[StrictStr]``, which raised a
    ValidationError on every ``get_logs`` call covering sign-on/DENY events.
    Behavior is deliberately unchanged from the original in-line patch: the same
    ``Optional[List[Any]]`` annotation and the same forced rebuild.  In
    particular the field's default is left alone — it is already optional, and
    this patch widens only the element type.
    """
    from okta.models.log_security_context import LogSecurityContext

    patched_type = t.Optional[t.List[t.Any]]
    LogSecurityContext.__annotations__["user_behaviors"] = patched_type
    model_fields = getattr(LogSecurityContext, "model_fields", {})
    if "user_behaviors" in model_fields:
        model_fields["user_behaviors"].annotation = patched_type
    LogSecurityContext.model_rebuild(force=True)
    return "relaxed userBehaviors to Optional[List[Any]]"


#: Ordered patch registry.  Each entry is ``(label, callable)``; the callable
#: returns a short description of what it did, logged at DEBUG.
_PATCHES: tuple[tuple[str, t.Callable[[], str]], ...] = (
    ("Defect A: SamlApplicationSettingsSignOn required booleans", _patch_defect_a),
    ("Defect B: narrow _embedded HAL envelope", _patch_defect_b),
    ("Defect C: closed authenticator key enum", _patch_defect_c),
    ("Defect D: required lists on policy condition models", _patch_defect_d),
    ("LogSecurityContext.user_behaviors", _patch_log_security_context),
)


def apply_okta_model_compat() -> None:
    """Apply every SDK model relaxation.  Idempotent and safe to call repeatedly.

    Must run **before** any SDK deserialization.  It is invoked at the top of
    :mod:`okta_mcp_server.server`, and every tool module does
    ``from okta_mcp_server.server import mcp``, which guarantees that ordering.

    Re-running is not merely harmless but useful: the subclass sweep picks up any
    SDK model class imported since the previous call.

    Each patch runs under its own guard, so a single failure — e.g. a future SDK
    release where a field no longer exists — logs a warning and leaves the
    remaining patches to apply normally.
    """
    for label, patch in _PATCHES:
        try:
            logger.debug(f"[okta_compat] {label}: {patch()}")
        except Exception as exc:  # noqa: BLE001 — one patch must not break the rest
            logger.warning(f"[okta_compat] could not apply patch '{label}': {exc}")