From bd45149832875503cd1920e91a21b001b8cf8876 Mon Sep 17 00:00:00 2001
From: Zhiwei Liang <zliang@akamai.com>
Date: Sat, 24 Oct 2026 16:19:29 -0400
Subject: [PATCH] TPT-4739: Add MFA enforcement support

---
 linode_api4/groups/account.py                 |  41 +++++
 linode_api4/objects/account.py                |  27 +++-
 linode_api4/objects/profile.py                |   8 +
 test/fixtures/account_settings.json           |   3 +-
 test/fixtures/account_tfa-optional-users.json |  11 ++
 test/fixtures/account_users_test-user.json    |   1 +
 test/fixtures/profile.json                    |   1 +
 .../models/account/test_account.py            |  76 +++++++++
 .../models/profile/test_profile.py            |   1 +
 test/unit/linode_client_test.py               | 149 +++++++++++++++++-
 test/unit/objects/account_test.py             |  41 +++++
 test/unit/objects/profile_test.py             |  47 ++++++
 12 files changed, 401 insertions(+), 5 deletions(-)
 create mode 100644 test/fixtures/account_tfa-optional-users.json

diff --git a/linode_api4/groups/account.py b/linode_api4/groups/account.py
index 6f8c6528e..840a390d2 100644
--- a/linode_api4/groups/account.py
+++ b/linode_api4/groups/account.py
@@ -17,6 +17,7 @@
     Payment,
     PaymentMethod,
     ServiceTransfer,
+    TFAOptionalUser,
     User,
 )
 
@@ -185,6 +186,46 @@ def users(self, *filters):
         """
         return self.client._get_and_filter(User, *filters)
 
+    def tfa_optional_users(self):
+        """
+        Lists users exempted from MFA enforcement, regardless of the current
+        account-wide policy, using ``GET /account/tfa-optional-users``.
+
+        :returns: References containing each exempted user's username and URL.
+        :rtype: PaginatedList of TFAOptionalUser
+        """
+        return self.client._get_and_filter(
+            TFAOptionalUser, endpoint="/account/tfa-optional-users"
+        )
+
+    def tfa_optional_users_update(self, usernames):
+        """
+        Replaces the entire list of users exempted from MFA enforcement using
+        ``PUT /account/tfa-optional-users``. Omitted users lose their exemptions.
+        This does not change the account-wide ``tfa_enforced`` setting.
+
+        :param usernames: Usernames to exempt. Pass an empty list to remove all
+                          exemptions.
+        :type usernames: list[str]
+
+        :returns: The API's ``tfa_optional_users_count`` and collection ``url``.
+        :rtype: MappedObject
+        :raises ApiError: If a username is invalid or access is denied.
+        """
+        result = self.client.put(
+            "/account/tfa-optional-users", data={"usernames": usernames}
+        )
+
+        if not all(
+            key in result for key in ("tfa_optional_users_count", "url")
+        ):
+            raise UnexpectedResponseError(
+                "Unexpected response when updating TFA optional users!",
+                json=result,
+            )
+
+        return MappedObject(**result)
+
     def logins(self):
         """
         Returns a collection of successful logins for all users on the account during the last 90 days.
diff --git a/linode_api4/objects/account.py b/linode_api4/objects/account.py
index a4aca1848..714856bd2 100644
--- a/linode_api4/objects/account.py
+++ b/linode_api4/objects/account.py
@@ -1,6 +1,8 @@
 from __future__ import annotations
 
+from dataclasses import dataclass
 from datetime import datetime
+from typing import Optional
 
 import requests
 from deprecated import deprecated
@@ -17,7 +19,7 @@
 from linode_api4.objects.networking import Firewall
 from linode_api4.objects.nodebalancer import NodeBalancer
 from linode_api4.objects.profile import PersonalAccessToken
-from linode_api4.objects.serializable import StrEnum
+from linode_api4.objects.serializable import JSONObject, StrEnum
 from linode_api4.objects.support import SupportTicket
 from linode_api4.objects.volume import Volume
 from linode_api4.objects.vpc import VPC
@@ -203,6 +205,10 @@ class AccountSettings(Base):
     """
     Information related to your Account settings.
 
+    ``tfa_enforced`` is the mutable account-wide default for MFA enforcement.
+    User exemptions are managed with :meth:`AccountGroup.tfa_optional_users_update`
+    and are retained when account-wide enforcement is disabled.
+
     API Documentation: https://techdocs.akamai.com/linode-api/reference/get-account-settings
     """
 
@@ -219,6 +225,7 @@ class AccountSettings(Base):
         "backups_enabled": Property(mutable=True),
         "interfaces_for_new_linodes": Property(mutable=True),
         "maintenance_policy": Property(mutable=True),
+        "tfa_enforced": Property(mutable=True),
     }
 
 
@@ -531,10 +538,27 @@ class Payment(Base):
     }
 
 
+@dataclass
+class TFAOptionalUser(JSONObject):
+    """
+    A reference to an account user exempted from MFA enforcement.
+
+    Returned by :meth:`AccountGroup.tfa_optional_users`. The ``username`` and
+    ``url`` fields identify the user; load a :class:`User` for full details.
+    """
+
+    username: Optional[str] = None
+    url: Optional[str] = None
+
+
 class User(Base):
     """
     An object representing a single user on this account.
 
+    ``tfa_enforced`` is the read-only effective MFA enforcement policy for this
+    user, distinct from whether MFA is enabled (``tfa_enabled``). It can be used
+    to filter :meth:`AccountGroup.users`.
+
     API Documentation: https://techdocs.akamai.com/linode-api/reference/get-user
     """
 
@@ -548,6 +572,7 @@ class User(Base):
         "restricted": Property(mutable=True),
         "ssh_keys": Property(),
         "tfa_enabled": Property(),
+        "tfa_enforced": Property(),
     }
 
     @property
diff --git a/linode_api4/objects/profile.py b/linode_api4/objects/profile.py
index c37015e84..b3b455841 100644
--- a/linode_api4/objects/profile.py
+++ b/linode_api4/objects/profile.py
@@ -60,6 +60,9 @@ class Profile(Base):
     """
     A Profile containing information about the current User.
 
+    ``tfa_enforced`` is the read-only effective MFA enforcement policy, distinct
+    from whether MFA is enabled (``two_factor_auth``).
+
     API Documentation: https://techdocs.akamai.com/linode-api/reference/get-profile
     """
 
@@ -77,6 +80,7 @@ class Profile(Base):
         "lish_auth_method": Property(mutable=True),
         "authorized_keys": Property(mutable=True),
         "two_factor_auth": Property(),
+        "tfa_enforced": Property(),
         "restricted": Property(),
         "authentication_type": Property(),
         "authorized_keys": Property(),
@@ -120,10 +124,14 @@ def disable_tfa(self):
         """
         Turns off TFA for this user's account.
 
+        The API rejects this operation when account-wide enforcement applies
+        and the user is not exempted.
+
         API Documentation: https://techdocs.akamai.com/linode-api/reference/post-tfa-disable
 
         :returns: Returns true if operation was successful
         :rtype: bool
+        :raises ApiError: If the API disallows disabling MFA.
         """
         self._client.post("/profile/tfa-disable")
 
diff --git a/test/fixtures/account_settings.json b/test/fixtures/account_settings.json
index 963c37306..80c9ad6c6 100644
--- a/test/fixtures/account_settings.json
+++ b/test/fixtures/account_settings.json
@@ -5,5 +5,6 @@
   "object_storage": "active",
   "backups_enabled": true,
   "interfaces_for_new_linodes": "linode_default_but_legacy_config_allowed",
-  "maintenance_policy": "linode/migrate"
+  "maintenance_policy": "linode/migrate",
+  "tfa_enforced": false
 }
diff --git a/test/fixtures/account_tfa-optional-users.json b/test/fixtures/account_tfa-optional-users.json
new file mode 100644
index 000000000..abfae1f04
--- /dev/null
+++ b/test/fixtures/account_tfa-optional-users.json
@@ -0,0 +1,11 @@
+{
+  "data": [
+    {
+      "username": "test-user",
+      "url": "/account/users/test-user"
+    }
+  ],
+  "page": 1,
+  "pages": 1,
+  "results": 1
+}
\ No newline at end of file
diff --git a/test/fixtures/account_users_test-user.json b/test/fixtures/account_users_test-user.json
index 66e5f9b12..7c8aa310c 100644
--- a/test/fixtures/account_users_test-user.json
+++ b/test/fixtures/account_users_test-user.json
@@ -6,5 +6,6 @@
       "laptop"
     ],
     "tfa_enabled": true,
+    "tfa_enforced": true,
     "username": "test-user"
   }
\ No newline at end of file
diff --git a/test/fixtures/profile.json b/test/fixtures/profile.json
index 2c62a70a0..9a2c3aaf5 100644
--- a/test/fixtures/profile.json
+++ b/test/fixtures/profile.json
@@ -18,6 +18,7 @@
     "restricted": false,
     "timezone": "US/Eastern",
     "two_factor_auth": true,
+    "tfa_enforced": false,
     "uid": 1234,
     "username": "exampleUser",
     "verified_phone_number": "+5555555555"
diff --git a/test/integration/models/account/test_account.py b/test/integration/models/account/test_account.py
index cb29f0de9..e183bdacf 100644
--- a/test/integration/models/account/test_account.py
+++ b/test/integration/models/account/test_account.py
@@ -9,6 +9,7 @@
 
 import pytest
 
+from linode_api4.errors import ApiError
 from linode_api4.objects import (
     Account,
     AccountSettings,
@@ -16,6 +17,7 @@
     ChildAccount,
     Event,
     Login,
+    TFAOptionalUser,
     User,
 )
 
@@ -71,6 +73,80 @@ def test_get_account_settings(test_linode_client):
     assert "maintenance_policy" in str(account_settings._raw_json)
 
 
+def test_account_tfa_enforced_round_trip(test_linode_client):
+    enforced = test_linode_client.account.settings().tfa_enforced
+    assert isinstance(enforced, bool)
+
+    settings = AccountSettings(test_linode_client, "")
+    settings.tfa_enforced = enforced
+
+    assert settings.save()
+    assert test_linode_client.account.settings().tfa_enforced is enforced
+
+
+def test_get_user_tfa_enforced(test_linode_client):
+    profile = test_linode_client.profile()
+    user = test_linode_client.load(User, profile.username)
+
+    assert isinstance(user.tfa_enforced, bool)
+    assert user.tfa_enforced is profile.tfa_enforced
+
+
+@pytest.mark.parametrize("enforced", [True, False])
+def test_list_users_tfa_enforced(test_linode_client, enforced):
+    profile = test_linode_client.profile()
+    users = list(
+        test_linode_client.account.users(User.tfa_enforced == enforced)
+    )
+
+    assert all(user.tfa_enforced is enforced for user in users)
+    if profile.tfa_enforced is enforced:
+        assert profile.username in {user.username for user in users}
+
+
+def test_tfa_optional_users_round_trip(test_linode_client):
+    client = test_linode_client
+    enforced = client.account.settings().tfa_enforced
+    usernames = sorted(
+        user.username for user in client.account.tfa_optional_users()
+    )
+
+    result = client.account.tfa_optional_users_update(usernames)
+
+    assert result.tfa_optional_users_count == len(usernames)
+    assert result.url in (
+        "/account/tfa-optional-users",
+        "/v4/account/tfa-optional-users",
+    )
+    users = list(client.account.tfa_optional_users())
+    assert sorted(user.username for user in users) == usernames
+    for user in users:
+        assert isinstance(user, TFAOptionalUser)
+        assert isinstance(user.username, str)
+        assert user.url in (
+            f"/account/users/{user.username}",
+            f"/v4/account/users/{user.username}",
+        )
+    assert client.account.settings().tfa_enforced is enforced
+
+
+def test_tfa_optional_users_invalid_username(test_linode_client):
+    client = test_linode_client
+    usernames = sorted(
+        user.username for user in client.account.tfa_optional_users()
+    )
+    invalid_username = f"missing-{time.time_ns()}"
+
+    with pytest.raises(ApiError) as error:
+        client.account.tfa_optional_users_update(usernames + [invalid_username])
+
+    assert error.value.status == 400
+    assert (
+        sorted(user.username for user in client.account.tfa_optional_users())
+        == usernames
+    )
+
+
 def test_update_maintenance_policy(test_linode_client):
     client = test_linode_client
     settings = client.load(AccountSettings(client, ""), "")
diff --git a/test/integration/models/profile/test_profile.py b/test/integration/models/profile/test_profile.py
index 6942eea38..5e289c7bd 100644
--- a/test/integration/models/profile/test_profile.py
+++ b/test/integration/models/profile/test_profile.py
@@ -10,6 +10,7 @@ def test_user_profile(test_linode_client):
     profile = client.profile()
 
     assert isinstance(profile, Profile)
+    assert isinstance(profile.tfa_enforced, bool)
 
 
 def test_get_personal_access_token_objects(test_linode_client):
diff --git a/test/unit/linode_client_test.py b/test/unit/linode_client_test.py
index c17be6006..3319215cb 100644
--- a/test/unit/linode_client_test.py
+++ b/test/unit/linode_client_test.py
@@ -1,7 +1,16 @@
+import json
+from copy import deepcopy
 from datetime import datetime
-from test.unit.base import ClientBaseCase
-
-from linode_api4 import FirewallCreateDevicesOptions, LongviewSubscription
+from test.unit.base import ClientBaseCase, load_json
+
+from linode_api4 import (
+    ApiError,
+    FirewallCreateDevicesOptions,
+    LongviewSubscription,
+    TFAOptionalUser,
+    UnexpectedResponseError,
+    User,
+)
 from linode_api4.objects.beta import BetaProgram
 from linode_api4.objects.linode import Instance
 from linode_api4.objects.networking import IPAddress
@@ -391,6 +400,140 @@ def test_get_settings(self):
         self.assertEqual(type(s.longview_subscription), LongviewSubscription)
         self.assertEqual(s.longview_subscription.id, "longview-100")
         self.assertEqual(s.object_storage, "active")
+        self.assertFalse(s.tfa_enforced)
+
+    def test_get_users_tfa_enforced_filter(self):
+        for enforced in (True, False):
+            with self.subTest(enforced=enforced):
+                user = deepcopy(load_json("/account/users/test-user"))
+                user["tfa_enforced"] = enforced
+                with self.mock_get(
+                    {"data": [user], "page": 1, "pages": 1, "results": 1}
+                ) as mock:
+                    users = self.client.account.users(
+                        User.tfa_enforced == enforced
+                    )
+
+                    self.assertEqual(mock.call_url, "/account/users")
+                    self.assertEqual(
+                        json.loads(mock.call_headers["X-Filter"]),
+                        {"tfa_enforced": enforced},
+                    )
+                    self.assertEqual(users[0].tfa_enforced, enforced)
+
+    def test_user_create_tfa_enforced(self):
+        for enforced in (True, False):
+            with self.subTest(enforced=enforced):
+                response = deepcopy(load_json("/account/users/test-user"))
+                response["tfa_enforced"] = enforced
+                with self.mock_post(response) as mock:
+                    user = self.client.account.user_create(
+                        response["email"], response["username"]
+                    )
+
+                    self.assertEqual(user.tfa_enforced, enforced)
+                    self.assertEqual(mock.call_url, "/account/users")
+                    self.assertEqual(
+                        mock.call_data,
+                        {
+                            "email": response["email"],
+                            "username": response["username"],
+                            "restricted": True,
+                        },
+                    )
+
+    def test_get_tfa_optional_users_pagination(self):
+        with self.mock_get("/account/tfa-optional-users") as mock:
+            first_page = deepcopy(mock.return_dct)
+            first_page.update(pages=2, results=2)
+            mock.mock.return_value._json = first_page
+            users = self.client.account.tfa_optional_users()
+            self.assertEqual(mock.call_url, "/account/tfa-optional-users")
+            self.assertEqual(mock.call_count, 1)
+            self.assertEqual(len(users), 2)
+            self.assertIsInstance(users[0], TFAOptionalUser)
+            self.assertEqual(users[0].username, "test-user")
+            self.assertEqual(users[0].url, "/account/users/test-user")
+
+            second_page = deepcopy(first_page)
+            second_page["page"] = 2
+            second_page["data"][0] = {
+                "username": "second-user",
+                "url": "/account/users/second-user",
+            }
+            mock.mock.return_value._json = second_page
+
+            self.assertEqual(users[1].username, "second-user")
+            self.assertEqual(users[1].url, "/account/users/second-user")
+            self.assertEqual(
+                mock.call_url,
+                "/account/tfa-optional-users?page=2&page_size=1",
+            )
+            self.assertEqual(mock.call_count, 2)
+
+    def test_get_tfa_optional_users_empty(self):
+        with self.mock_get({"data": [], "page": 1, "pages": 1, "results": 0}):
+            self.assertEqual(list(self.client.account.tfa_optional_users()), [])
+
+    def test_update_tfa_optional_users(self):
+        for usernames in (["user1", "user2"], []):
+            with self.subTest(usernames=usernames):
+                response = {
+                    "tfa_optional_users_count": len(usernames),
+                    "url": "/account/tfa-optional-users",
+                }
+                with self.mock_put(response) as mock:
+                    result = self.client.account.tfa_optional_users_update(
+                        usernames
+                    )
+
+                    self.assertEqual(
+                        mock.call_url, "/account/tfa-optional-users"
+                    )
+                    self.assertEqual(mock.call_data, {"usernames": usernames})
+                    self.assertEqual(
+                        result.tfa_optional_users_count, len(usernames)
+                    )
+                    self.assertEqual(result.url, response["url"])
+
+    def test_update_tfa_optional_users_unexpected_response(self):
+        for response in (
+            {},
+            {"url": "/account/tfa-optional-users"},
+            {"tfa_optional_users_count": 0},
+        ):
+            with self.subTest(response=response):
+                with self.mock_put(response):
+                    with self.assertRaises(UnexpectedResponseError):
+                        self.client.account.tfa_optional_users_update([])
+
+    def test_update_tfa_optional_users_api_error(self):
+        response = {"errors": [{"reason": "Invalid user or access denied."}]}
+        for status_code in (400, 403):
+            with self.subTest(status_code=status_code):
+                with self.mock_put(response) as mock:
+                    mock.mock.return_value.status_code = status_code
+                    mock.mock.return_value.request = None
+
+                    with self.assertRaises(ApiError) as raised:
+                        self.client.account.tfa_optional_users_update(
+                            ["test-user"]
+                        )
+
+                    self.assertEqual(raised.exception.status, status_code)
+                    self.assertEqual(raised.exception.json, response)
+
+    def test_get_tfa_optional_users_access_denied(self):
+        response = {"errors": [{"reason": "Access denied."}]}
+        with self.mock_get(response) as mock:
+            mock.mock.return_value.status_code = 403
+            mock.mock.return_value.request = None
+
+            with self.assertRaises(ApiError) as raised:
+                self.client.account.tfa_optional_users()
+
+            self.assertEqual(raised.exception.status, 403)
+            self.assertEqual(raised.exception.json, response)
 
     def test_get_invoices(self):
         """
diff --git a/test/unit/objects/account_test.py b/test/unit/objects/account_test.py
index da807d182..62c9ea83c 100644
--- a/test/unit/objects/account_test.py
+++ b/test/unit/objects/account_test.py
@@ -123,6 +123,7 @@ def test_get_account_settings(self):
         self.assertEqual(settings.network_helper, False)
         self.assertEqual(settings.object_storage, "active")
         self.assertEqual(settings.backups_enabled, True)
+        self.assertFalse(settings.tfa_enforced)
         self.assertEqual(
             settings.interfaces_for_new_linodes,
             AccountSettingsInterfacesForNewLinodes.linode_default_but_legacy_config_allowed,
@@ -148,6 +149,7 @@ def test_post_account_settings(self):
                 "backups_enabled": False,
                 "interfaces_for_new_linodes": AccountSettingsInterfacesForNewLinodes.linode_only,
                 "maintenance_policy": "linode/migrate",
+                "tfa_enforced": False,
             }
 
     def test_update_account_settings(self):
@@ -168,6 +170,23 @@ def test_update_account_settings(self):
                 },
             )
 
+    def test_update_account_settings_tfa_enforced(self):
+        for enforced in (True, False):
+            with self.subTest(enforced=enforced):
+                settings = AccountSettings(
+                    self.client,
+                    False,
+                    {"managed": False, "tfa_enforced": not enforced},
+                )
+                self.assertEqual(settings.tfa_enforced, not enforced)
+                settings.tfa_enforced = enforced
+
+                with self.mock_put({}) as mock:
+                    settings.save()
+
+                    self.assertEqual(mock.call_url, "/account/settings")
+                    self.assertEqual(mock.call_data, {"tfa_enforced": enforced})
+
     def test_get_event(self):
         """
         Tests that an event is loaded correctly by ID
@@ -252,8 +271,30 @@ def test_get_user(self):
         self.assertEqual(user.email, "test-user@linode.com")
         self.assertTrue(user.restricted)
         self.assertTrue(user.tfa_enabled)
+        self.assertTrue(user.tfa_enforced)
         self.assertIsNotNone(user.ssh_keys)
 
+    def test_user_tfa_enforced_read_only(self):
+        for enforced in (True, False):
+            with self.subTest(enforced=enforced):
+                user = User(
+                    self.client,
+                    "test-user",
+                    {"username": "test-user", "tfa_enforced": enforced},
+                )
+                self.assertEqual(user.tfa_enforced, enforced)
+                with self.assertRaises(AttributeError):
+                    user.tfa_enforced = not enforced
+
+                user.restricted = False
+                with self.mock_put({}) as mock:
+                    user.save()
+                    self.assertEqual(mock.call_url, "/account/users/test-user")
+                    self.assertEqual(
+                        mock.call_data,
+                        {"username": "test-user", "restricted": False},
+                    )
+
     def test_get_service_transfer(self):
         """
         Tests that a service transfer is loaded correctly by token
diff --git a/test/unit/objects/profile_test.py b/test/unit/objects/profile_test.py
index cbe8dabd7..427b76e46 100644
--- a/test/unit/objects/profile_test.py
+++ b/test/unit/objects/profile_test.py
@@ -1,6 +1,7 @@
 from datetime import datetime
 from test.unit.base import ClientBaseCase
 
+from linode_api4 import ApiError
 from linode_api4.objects import Profile, ProfileLogin, SSHKey
 from linode_api4.objects.profile import TrustedDevice
 
@@ -91,9 +92,55 @@ def test_get_profile(self):
         self.assertFalse(profile.restricted)
         self.assertEqual(profile.timezone, "US/Eastern")
         self.assertTrue(profile.two_factor_auth)
+        self.assertFalse(profile.tfa_enforced)
         self.assertEqual(profile.uid, 1234)
         self.assertEqual(profile.verified_phone_number, "+5555555555")
 
+    def test_profile_tfa_enforced_read_only(self):
+        for enforced in (True, False):
+            with self.subTest(enforced=enforced):
+                profile = Profile(
+                    self.client,
+                    "exampleUser",
+                    {"username": "exampleUser", "tfa_enforced": enforced},
+                )
+                self.assertEqual(profile.tfa_enforced, enforced)
+                with self.assertRaises(AttributeError):
+                    profile.tfa_enforced = not enforced
+
+                profile.email_notifications = False
+                with self.mock_put({}) as mock:
+                    profile.save()
+                    self.assertEqual(mock.call_url, "/profile")
+                    self.assertEqual(
+                        mock.call_data, {"email_notifications": False}
+                    )
+
+    def test_disable_tfa(self):
+        profile = Profile(self.client, "exampleUser")
+
+        with self.mock_post({}) as mock:
+            self.assertTrue(profile.disable_tfa())
+            self.assertEqual(mock.call_url, "/profile/tfa-disable")
+
+    def test_disable_tfa_enforced(self):
+        profile = Profile(
+            self.client,
+            "exampleUser",
+            {"username": "exampleUser", "tfa_enforced": True},
+        )
+        response = {"errors": [{"reason": "MFA is enforced for this user."}]}
+        with self.mock_post(response) as mock:
+            mock.mock.return_value.status_code = 400
+            mock.mock.return_value.request = None
+
+            with self.assertRaises(ApiError) as raised:
+                profile.disable_tfa()
+
+            self.assertEqual(mock.call_url, "/profile/tfa-disable")
+            self.assertEqual(raised.exception.status, 400)
+            self.assertEqual(raised.exception.json, response)
+
     def test_get_trusted_device(self):
         device = TrustedDevice(self.client, 123)