diff --git a/cloudinary/api_client/call_account_api.py b/cloudinary/api_client/call_account_api.py index 7e1f4020..06c1726e 100644 --- a/cloudinary/api_client/call_account_api.py +++ b/cloudinary/api_client/call_account_api.py @@ -1,7 +1,7 @@ import cloudinary from cloudinary.api_client.execute_request import execute_request from cloudinary.provisioning.account_config import account_config -from cloudinary.utils import get_http_connector, normalize_params +from cloudinary.utils import get_http_connector, json_body, normalize_params PROVISIONING_SUB_PATH = "provisioning" ACCOUNT_SUB_PATH = "accounts" @@ -46,9 +46,15 @@ def _execute_account_request(method, uri, auth, params=None, headers=None, **opt api_version = options.pop("api_version", cloudinary.API_VERSION) provisioning_api_url = "/".join([prefix, api_version, PROVISIONING_SUB_PATH] + uri) + params = normalize_params(params) + + if method.upper() != "GET": + options["body"], headers = json_body(params, headers) + params = None + return execute_request(http_connector=_http, method=method, - params=normalize_params(params), + params=params, headers=headers, auth=auth, api_url=provisioning_api_url, diff --git a/cloudinary/api_client/call_api.py b/cloudinary/api_client/call_api.py index 0e93b33c..39131caa 100644 --- a/cloudinary/api_client/call_api.py +++ b/cloudinary/api_client/call_api.py @@ -2,7 +2,7 @@ import cloudinary from cloudinary.api_client.execute_request import execute_request -from cloudinary.utils import get_http_connector, normalize_params +from cloudinary.utils import get_http_connector, json_body, normalize_params logger = cloudinary.logger _http = get_http_connector(cloudinary.config(), cloudinary.CERT_KWARGS) @@ -35,12 +35,13 @@ def call_metadata_rules_api(method, uri, params, **options): def call_json_api(method, uri, params, **options): - data=None + data = None + headers = {'Content-Type': 'application/json'} if method.upper() != 'GET': - data = json.dumps(params).encode('utf-8') + data, headers = json_body(params) params = None - return _call_api(method, uri, params=params, body=data, headers={'Content-Type': 'application/json'}, **options) + return _call_api(method, uri, params=params, body=data, headers=headers, **options) def _call_v2_api(method, uri, params, **options): diff --git a/cloudinary/provisioning/__init__.py b/cloudinary/provisioning/__init__.py index 1f8005c6..55045ca3 100644 --- a/cloudinary/provisioning/__init__.py +++ b/cloudinary/provisioning/__init__.py @@ -1,5 +1,5 @@ from .account_config import AccountConfig, account_config, reset_config -from .account import (create_agent_account, +from .account import (create_agent_account, create_cloud, sub_accounts, create_sub_account, delete_sub_account, sub_account, update_sub_account, user_groups, create_user_group, update_user_group, delete_user_group, user_group, add_user_to_group, remove_user_from_group, user_group_users, user_in_user_groups, diff --git a/cloudinary/provisioning/account.py b/cloudinary/provisioning/account.py index f54382a0..9b970d25 100644 --- a/cloudinary/provisioning/account.py +++ b/cloudinary/provisioning/account.py @@ -2,6 +2,7 @@ from cloudinary.utils import encode_list AGENTS_SUB_PATH = "agents" +CLOUDS_SUB_PATH = "clouds" SUB_ACCOUNTS_SUB_PATH = "sub_accounts" USERS_SUB_PATH = "users" USER_GROUPS_SUB_PATH = "user_groups" @@ -60,6 +61,61 @@ def create_agent_account(email, agent_framework, agent_llm_model, agent_goal, sd return _call_public_account_api("POST", uri, params=params, **options) +def create_cloud(delivery_ips=None, email=None, agent_framework=None, agent_llm_model=None, agent_goal=None, + sdk_framework=None, **options): + """ + Create a Claimable Cloud, intended for use by AI agents. + + Creates a temporary cloud whose credentials work immediately, with media delivery restricted + to an IP allow-list. No verification email is sent. Unless a human claims it via the returned + claim_url, the cloud is disabled when it expires; claiming makes it permanent, keeps the + credentials and assets, and lifts the IP restriction. + + Creation only: a cloud cannot be read, updated or deleted, so the delivery IPs are fixed for + its lifetime. If they are wrong, create another cloud or claim this one. + + The restriction covers media delivery only, not the Upload and Admin APIs, so it is not a + confidentiality control. Uploads succeeding while delivery fails with `x-cld-error: ACL deny` + is the expected symptom of it. + + This endpoint is public and unauthenticated, and is rate-limited per IP address. + + :param delivery_ips: Up to three additional IP addresses permitted to deliver media, for + hosts other than the caller. The caller's own resolved address is + always appended, so omitting this is the usual call; the literal + "requester_ip" is replaced by that address. IPv4 and IPv6 are + accepted, CIDR ranges are not. Non-public addresses are dropped and + the call fails unless at least one public address remains, so read + delivery_ips back from the response rather than assuming the list + sent was stored. + :type delivery_ips: list[str], optional + :param email: Email address to associate the claim with. Not verified, and no mail + is sent to it; a placeholder is generated when omitted. + :type email: str, optional + :param agent_framework: The name of the agent framework used to create the cloud. + :type agent_framework: str, optional + :param agent_llm_model: The LLM model powering the agent. + :type agent_llm_model: str, optional + :param agent_goal: A short description of what the agent is trying to achieve. + :type agent_goal: str, optional + :param sdk_framework: The Cloudinary SDK framework the agent intends to use. + :type sdk_framework: str, optional + :param options: Generic advanced options dict, see online documentation + :type options: dict, optional + :return: The created Claimable Cloud, including working credentials, + the claim URL and the expiry time + :rtype: dict + """ + uri = [CLOUDS_SUB_PATH] + params = {"delivery_ips": delivery_ips, + "email": email, + "agent_framework": agent_framework, + "agent_llm_model": agent_llm_model, + "agent_goal": agent_goal, + "sdk_framework": sdk_framework} + return _call_public_account_api("POST", uri, params=params, **options) + + def sub_accounts(enabled=None, ids=None, prefix=None, **options): """ List all sub accounts @@ -163,7 +219,7 @@ def update_sub_account(sub_account_id, name=None, cloud_name=None, custom_attrib return _call_account_api("put", uri, params=params, **options) -def users(user_ids=None, sub_account_id=None, pending=None, prefix=None, last_login=None, from_date=None, to_date=None, +def users(user_ids=None, sub_account_id=None, status=None, prefix=None, last_login=None, from_date=None, to_date=None, **options): """ List all users @@ -171,10 +227,10 @@ def users(user_ids=None, sub_account_id=None, pending=None, prefix=None, last_lo :type user_ids: list, optional :param sub_account_id: The id of a sub account :type sub_account_id: str, optional - :param pending: Limit results to pending users (True), - users that are not pending (False), - or all users (None, the default). - :type pending: bool, optional + :param status: Limit results to users of this status: "pending" for users who have not yet + set a password, otherwise a user status such as "active". All users when + omitted. + :type status: str, optional :param prefix: User prefix :type prefix: str, optional :param last_login: Return only users that last logged in in the specified range of dates (true), @@ -193,7 +249,7 @@ def users(user_ids=None, sub_account_id=None, pending=None, prefix=None, last_lo user_ids = encode_list(user_ids) params = {"ids": user_ids, "sub_account_id": sub_account_id, - "pending": pending, + "status": status, "prefix": prefix, "last_login": last_login, "from": from_date, diff --git a/cloudinary/utils.py b/cloudinary/utils.py index be21fb79..26876d3e 100644 --- a/cloudinary/utils.py +++ b/cloudinary/utils.py @@ -613,6 +613,20 @@ def normalize_params(params): return dict([(k, __bool_string(v)) for (k, v) in params.items() if v is not None and not v == ""]) +def json_body(params, headers=None): + """ + Encodes params as a JSON request body with the matching Content-Type. + + :param params: Params to serialize. + :param headers: Headers to extend. A Content-Type already present is kept. + :return: Tuple of the encoded body and the resulting headers. + """ + headers = dict(headers or {}) + headers.setdefault("Content-Type", "application/json") + + return json.dumps(params).encode("utf-8"), headers + + def sign_request(params, options): api_key = options.get("api_key", cloudinary.config().api_key) if not api_key: diff --git a/test/test_provisioning_api.py b/test/test_provisioning_api.py index 80f88f3b..e5f1e7ad 100644 --- a/test/test_provisioning_api.py +++ b/test/test_provisioning_api.py @@ -7,10 +7,10 @@ import cloudinary.provisioning.account from cloudinary.provisioning import account_config, reset_config -from cloudinary.exceptions import AuthorizationRequired, NotFound +from cloudinary.exceptions import AuthorizationRequired, BadRequest, NotFound, RateLimited -from test.helper_test import (UNIQUE_SUB_ACCOUNT_ID, UNIQUE_TEST_ID, URLLIB3_REQUEST, patch, api_response_mock, - get_uri, get_method, get_params, get_headers) +from test.helper_test import (SUFFIX, UNIQUE_SUB_ACCOUNT_ID, UNIQUE_TEST_ID, URLLIB3_REQUEST, patch, + api_response_mock, http_response_mock, get_uri, get_method, get_params, get_headers) disable_warnings() @@ -22,7 +22,7 @@ class AccountApiTest(unittest.TestCase): @classmethod def setUpClass(cls): - now = datetime.now().strftime("%m-%d-%Y") + now = "{0}-{1}".format(datetime.now().strftime("%m-%d-%Y"), SUFFIX) cls.user_name_1 = "SDK TEST " + now cls.user_name_2 = "SDK TEST 2 " + now user_email_1 = "sdk-test" + now + "@cloudinary.com" @@ -103,7 +103,7 @@ def test_get_specific_sub_account(self): @unittest.skipUnless(cloudinary.provisioning.account_config().provisioning_api_secret, "requires provisioning_api_key/provisioning_api_secret") def test_update_user(self): - now = datetime.now().strftime("%m-%d-%Y") + now = "{0}-{1}".format(datetime.now().strftime("%m-%d-%Y"), SUFFIX) new_email_address = "updated" + now + "@cloudinary.com" new_name = "updated" @@ -129,56 +129,56 @@ def test_get_users(self): @unittest.skipUnless(cloudinary.provisioning.account_config().provisioning_api_secret, "requires provisioning_api_key/provisioning_api_secret") def test_get_pending_users(self): - res = cloudinary.provisioning.users(user_ids=[self.user_id_1], pending=True) + res = cloudinary.provisioning.users(user_ids=[self.user_id_1], status="pending") self.assertEqual(len(res["users"]), 1) @unittest.skipUnless(cloudinary.provisioning.account_config().provisioning_api_secret, "requires provisioning_api_key/provisioning_api_secret") def test_get_non_pending_users(self): - res = cloudinary.provisioning.users(user_ids=[self.user_id_1], pending=False) + res = cloudinary.provisioning.users(user_ids=[self.user_id_1], status="active") self.assertEqual(len(res["users"]), 0) @unittest.skipUnless(cloudinary.provisioning.account_config().provisioning_api_secret, "requires provisioning_api_key/provisioning_api_secret") def test_get_pending_and_non_pending_users(self): - res = cloudinary.provisioning.users(user_ids=[self.user_id_1], pending=None) + res = cloudinary.provisioning.users(user_ids=[self.user_id_1], status=None) self.assertEqual(len(res["users"]), 1) @unittest.skipUnless(cloudinary.provisioning.account_config().provisioning_api_secret, "requires provisioning_api_key/provisioning_api_secret") def test_get_users_by_prefix(self): - res_1 = cloudinary.provisioning.users(pending=True, prefix=self.user_name_2[:-1]) - res_2 = cloudinary.provisioning.users(pending=True, prefix=self.user_name_2+'zzz') + res_1 = cloudinary.provisioning.users(status="pending", prefix=self.user_name_2[:-1]) + res_2 = cloudinary.provisioning.users(status="pending", prefix=self.user_name_2+'zzz') self.assertEqual(len(res_1["users"]), 1) self.assertEqual(len(res_2["users"]), 0) @unittest.skipUnless(cloudinary.provisioning.account_config().provisioning_api_secret, "requires provisioning_api_key/provisioning_api_secret") def test_get_users_by_sub_account_id(self): - res = cloudinary.provisioning.users(pending=True, user_ids=[self.user_id_2], sub_account_id=self.cloud_id) + res = cloudinary.provisioning.users(status="pending", user_ids=[self.user_id_2], sub_account_id=self.cloud_id) self.assertEqual(len(res["users"]), 1) @unittest.skipUnless(cloudinary.provisioning.account_config().provisioning_api_secret, "requires provisioning_api_key/provisioning_api_secret") def test_get_users_by_nonexistent_sub_account_id(self): with six.assertRaisesRegex(self, NotFound, "Cannot find sub account with id {}".format(UNIQUE_SUB_ACCOUNT_ID)): - cloudinary.provisioning.users(pending=True, sub_account_id=UNIQUE_SUB_ACCOUNT_ID) + cloudinary.provisioning.users(status="pending", sub_account_id=UNIQUE_SUB_ACCOUNT_ID) @unittest.skipUnless(cloudinary.provisioning.account_config().provisioning_api_secret, "requires provisioning_api_key/provisioning_api_secret") def test_get_users_by_login(self): - res = cloudinary.provisioning.users(user_ids=[self.user_id_1], pending=None, + res = cloudinary.provisioning.users(user_ids=[self.user_id_1], status=None, last_login="true", from_date=datetime.today(), to_date=datetime.today()) self.assertEqual(len(res["users"]), 0) - res = cloudinary.provisioning.users(user_ids=[self.user_id_1], pending=None, + res = cloudinary.provisioning.users(user_ids=[self.user_id_1], status=None, last_login="false", from_date=datetime.today(), to_date=datetime.today()) self.assertEqual(len(res["users"]), 1) @unittest.skipUnless(cloudinary.provisioning.account_config().provisioning_api_secret, "requires provisioning_api_key/provisioning_api_secret") def test_update_user_group(self): - now = datetime.now().strftime("%m-%d-%Y") + now = "{0}-{1}".format(datetime.now().strftime("%m-%d-%Y"), SUFFIX) new_name = "new-test-name" + now res = cloudinary.provisioning.update_user_group(self.group_id, new_name) self.assertEqual(res["id"], self.group_id) @@ -355,5 +355,181 @@ def test_create_agent_account_parses_response(self): self.assertIn("guidance", res) +class CreateCloudTest(unittest.TestCase): + """ + The create cloud endpoint is public, unauthenticated and rate limited per IP, and every + successful call provisions a real account, so it is verified against a mocked transport + rather than the live API. + """ + + def test_create_cloud(self): + with patch(URLLIB3_REQUEST) as mocker: + mocker.return_value = api_response_mock() + cloudinary.provisioning.create_cloud() + + self.assertEqual("POST", get_method(mocker)) + uri = get_uri(mocker) + self.assertTrue(uri.endswith("/provisioning/clouds")) + # The resource sits directly under provisioning/, with no agents/ prefix and no + # accounts/{account_id} segment. + self.assertNotIn("/agents", uri) + self.assertNotIn("/accounts", uri) + + def test_create_cloud_is_unauthenticated(self): + with patch(URLLIB3_REQUEST) as mocker: + mocker.return_value = api_response_mock() + cloudinary.provisioning.create_cloud() + + # The endpoint is public - no authorization header must be sent. + headers = get_headers(mocker) + self.assertNotIn("authorization", {k.lower() for k in headers}) + + def test_create_cloud_omits_unset_delivery_ips(self): + with patch(URLLIB3_REQUEST) as mocker: + mocker.return_value = api_response_mock() + cloudinary.provisioning.create_cloud() + + # The default path sends no delivery_ips at all, letting the server derive the + # allow-list from the requester's own resolved address. + self.assertNotIn("delivery_ips", get_params(mocker)) + # Nothing else is sent either - an empty body is the documented default request. + self.assertEqual({}, get_params(mocker)) + + def test_create_cloud_sends_requester_ip_sentinel(self): + with patch(URLLIB3_REQUEST) as mocker: + mocker.return_value = api_response_mock() + cloudinary.provisioning.create_cloud(["requester_ip"]) + + # Passed through verbatim; the server substitutes its own resolved address. + self.assertEqual(["requester_ip"], get_params(mocker)["delivery_ips"]) + + def test_create_cloud_sends_delivery_ips_as_array(self): + with patch(URLLIB3_REQUEST) as mocker: + mocker.return_value = api_response_mock() + cloudinary.provisioning.create_cloud(["8.8.8.8", "1.1.1.1", "requester_ip"]) + + # The server rejects a comma-joined string with "delivery_ips must be an array of + # IP addresses", so the list must stay a genuine array on the wire. + self.assertEqual(["8.8.8.8", "1.1.1.1", "requester_ip"], get_params(mocker)["delivery_ips"]) + + def test_create_cloud_sends_optional_email(self): + with patch(URLLIB3_REQUEST) as mocker: + mocker.return_value = api_response_mock() + cloudinary.provisioning.create_cloud(email="jane@example.com") + + self.assertEqual("jane@example.com", get_params(mocker)["email"]) + + def test_create_cloud_omits_unset_email(self): + with patch(URLLIB3_REQUEST) as mocker: + mocker.return_value = api_response_mock() + cloudinary.provisioning.create_cloud(["8.8.8.8"]) + + # Omitted rather than sent empty: the server generates a placeholder address. + self.assertNotIn("email", get_params(mocker)) + + def test_create_cloud_sends_optional_agent_metadata(self): + with patch(URLLIB3_REQUEST) as mocker: + mocker.return_value = api_response_mock() + cloudinary.provisioning.create_cloud( + agent_framework="langchain", + agent_llm_model="claude-opus-5", + agent_goal="Build a product image gallery", + sdk_framework="python", + ) + + params = get_params(mocker) + self.assertEqual("langchain", params["agent_framework"]) + self.assertEqual("claude-opus-5", params["agent_llm_model"]) + self.assertEqual("Build a product image gallery", params["agent_goal"]) + self.assertEqual("python", params["sdk_framework"]) + + def test_create_cloud_omits_unset_agent_metadata(self): + with patch(URLLIB3_REQUEST) as mocker: + mocker.return_value = api_response_mock() + cloudinary.provisioning.create_cloud() + + params = get_params(mocker) + for field in ("agent_framework", "agent_llm_model", "agent_goal", "sdk_framework"): + self.assertNotIn(field, params) + + def test_create_cloud_parses_response(self): + body = json.dumps({ + "account_id": "00000000-0000-0000-0000-000000000000", + "email": "cloud-0000000000000000@cloud.cloudinary.invalid", + "cloud_name": "test-cloud", + "api_key": "000000000000000", + "api_secret": "FAKE_API_SECRET_FOR_TESTS", + "api_environment_variable": + "CLOUDINARY_URL=cloudinary://000000000000000:FAKE_API_SECRET_FOR_TESTS@test-cloud", + "claimed": False, + "expires_at": "2026-08-13T13:08:42Z", + "delivery_ips": ["8.8.8.8"], + "claim_url": "https://console.cloudinary.com/users/agent_email_confirmation?token=FAKE_CLAIM_TOKEN", + "guidance": "A Claimable Cloud is ready and the API key and secret below work immediately.", + }) + with patch(URLLIB3_REQUEST) as mocker: + mocker.return_value = api_response_mock(body) + res = cloudinary.provisioning.create_cloud(["8.8.8.8"]) + + self.assertEqual("00000000-0000-0000-0000-000000000000", res["account_id"]) + self.assertEqual("test-cloud", res["cloud_name"]) + self.assertEqual("000000000000000", res["api_key"]) + self.assertEqual("FAKE_API_SECRET_FOR_TESTS", res["api_secret"]) + self.assertIn("CLOUDINARY_URL=cloudinary://", res["api_environment_variable"]) + self.assertFalse(res["claimed"]) + self.assertEqual("2026-08-13T13:08:42Z", res["expires_at"]) + self.assertEqual(["8.8.8.8"], res["delivery_ips"]) + self.assertIn("agent_email_confirmation", res["claim_url"]) + self.assertIn("guidance", res) + + def test_create_cloud_passes_through_unknown_response_shape(self): + # The contracted response is flat, but it is returned verbatim rather than reshaped, + # so unrecognized or added fields (here credentials nested under + # product_environments[], the shape the agent-account endpoint uses) still reach the + # caller intact instead of being dropped. + body = json.dumps({ + "id": "00000000-0000-0000-0000-000000000000", + "email": "cloud-0000000000000000@cloud.cloudinary.invalid", + "expires_at": "2026-08-13T13:08:42Z", + "delivery_ips": ["8.8.8.8"], + "claim_url": "https://console.cloudinary.com/users/agent_email_confirmation?token=FAKE_CLAIM_TOKEN", + "product_environments": [{ + "cloud_name": "test-cloud", + "api_access_keys": [{"key": "000000000000000", + "secret": "FAKE_API_SECRET_FOR_TESTS", + "enabled": True}], + }], + }) + with patch(URLLIB3_REQUEST) as mocker: + mocker.return_value = api_response_mock(body) + res = cloudinary.provisioning.create_cloud(["8.8.8.8"]) + + product_environment = res["product_environments"][0] + self.assertEqual("test-cloud", product_environment["cloud_name"]) + self.assertEqual("000000000000000", product_environment["api_access_keys"][0]["key"]) + self.assertEqual("FAKE_API_SECRET_FOR_TESTS", product_environment["api_access_keys"][0]["secret"]) + + def test_create_cloud_maps_errors(self): + for status, code in ((400, "delivery_ips_not_public"), + (400, "delivery_ips_invalid"), + (400, "delivery_ips_too_many")): + body = json.dumps({"error": {"category": "invalid_parameter", + "code": code, + "message": "delivery_ips error"}}) + with patch(URLLIB3_REQUEST) as mocker: + mocker.return_value = http_response_mock(body, status=status) + with self.assertRaises(BadRequest): + cloudinary.provisioning.create_cloud(["not-an-ip"]) + + for status in (420, 429): + body = json.dumps({"error": {"category": "rate_limit", + "code": "ip_rate_limit_exceeded", + "message": "Rate limit exceeded"}}) + with patch(URLLIB3_REQUEST) as mocker: + mocker.return_value = http_response_mock(body, status=status) + with self.assertRaises(RateLimited): + cloudinary.provisioning.create_cloud(["8.8.8.8"]) + + if __name__ == '__main__': unittest.main()