23051079501

Committed 13 Mar 2026 12:35PM UTC coverage: 94.545% (-0.1%) from 94.661%

Build # 23051079501

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #80 from chift-oneapi/nathan/fix_auth_error_bypass

fix(http): fix exception error_code propagation when the status is 401

Coverage Stats

0 of 6 new or added lines in 1 file covered. (0.0%)

1 existing line in 1 file now uncovered.

4645 of 4913 relevant lines covered (94.55%)

0.95 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

52.1

/chift/api/client.py

import base64
import http.client as httplib
import json
from datetime import datetime

import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

from chift.api import exceptions


class ChiftAuth(requests.auth.AuthBase):
    def __init__(
        self, client_id, client_secret, account_id, url_base, env_id, test_client
    ):
        self.client_id = client_id
        self.client_secret = client_secret
        self.account_id = account_id
        self.url_base = url_base
        self.env_id = env_id
        self.test_client = test_client
        self.request_engine = self.test_client or requests

        self.access_token = None
        self.expires_at = None

    def __call__(self, request):
        request.headers.update(self.get_auth_header())
        return request

    def get_auth_header(self):
        return {"Authorization": f"Bearer {self.get_access_token()}"}

    def _parse_token(self, token):
        self.access_token = token.get("access_token")
        self.expires_at = datetime.fromtimestamp(token.get("expires_on"))

    def get_access_token(self):
        if self.access_token:
            if datetime.now() < self.expires_at:
                return self.access_token

        payload = {
            "clientId": self.client_id,
            "clientSecret": self.client_secret,
            "accountId": self.account_id,
        }
        if self.env_id:
            payload["envId"] = self.env_id

        response = self.request_engine.post(self.url_base + "/token", json=payload)

        if not response.status_code == httplib.OK:
            raise exceptions.ChiftException(
                f"Error while authenticating '{response.status_code}': {response.text}"
            )

        self._parse_token(response.json())

        return self.access_token


class ChiftClient:
    session = None
    access_token = None
    auth = None
    consumer_id = None
    connection_id = None
    raw_data = None
    client_request_id = None
    related_chain_execution_id = None
    sync_id = None

    __instance = None
    __use_global = False

    def __new__(cls, **kwargs):
        """
        __use_global=True for singleton usage
        """
        if ChiftClient.__use_global:
            if ChiftClient.__instance is None:
                ChiftClient.__instance = object.__new__(cls)
            instance = ChiftClient.__instance
        else:
            instance = object.__new__(cls)

        return instance

    def __init__(self, consumer_id=None, **kwargs):
        from chift import (
            account_id,
            client_id,
            client_secret,
            related_chain_execution_id,
            sync_id,
            url_base,
        )

        self.consumer_id = consumer_id
        self.client_id = kwargs.get("client_id") or client_id
        self.client_secret = kwargs.get("client_secret") or client_secret
        self.account_id = kwargs.get("account_id") or account_id
        self.url_base = kwargs.get("url_base") or url_base
        self.env_id = kwargs.get("env_id")
        self.test_client = kwargs.get("test_client")
        self.related_chain_execution_id = (
            kwargs.get("related_chain_execution_id") or related_chain_execution_id
        )
        self.sync_id = kwargs.get("sync_id") or sync_id
        self.max_retries = kwargs.get("max_retries")
        self._start_session()

    def _start_session(self):
        self.ignored_error_codes = []
        self.extra_context = None
        if self.test_client:
            self.url_base = ""  # set to empty string to avoid url_base in the request
            self.test_auth = ChiftAuth(
                self.client_id,
                self.client_secret,
                self.account_id,
                self.url_base,
                self.env_id,
                self.test_client,
            )
        elif not self.session:
            self.session = requests.Session()
            self.session.auth = ChiftAuth(
                self.client_id,
                self.client_secret,
                self.account_id,
                self.url_base,
                self.env_id,
                None,
            )

            if self.max_retries:
                retries = Retry(
                    total=int(self.max_retries),
                    backoff_factor=0.1,
                    status_forcelist=[502],
                )
                self.session.mount("http://", HTTPAdapter(max_retries=retries))
                self.session.mount("https://", HTTPAdapter(max_retries=retries))

    def make_request(
        self, request_type, url, data=None, content_type="application/json", params=None
    ):
        if not params:
            params = {}

        if isinstance(data, str):
            data = json.loads(data)

        headers = {
            "Content-Type": content_type,
            "Accept": "application/json",
            "User-Agent": "chift-python-sdk library",
            "Accept-Encoding": "gzip",
        }

        if self.connection_id:
            headers["x-chift-connectionid"] = self.connection_id

        if self.raw_data:
            headers["x-chift-raw-data"] = "true"

        if self.client_request_id:
            headers["x-chift-client-requestid"] = self.client_request_id

        if self.related_chain_execution_id:
            headers["x-chift-relatedchainexecutionid"] = self.related_chain_execution_id

        if self.sync_id:
            headers["x-chift-syncid"] = self.sync_id

        if self.ignored_error_codes:
            headers["x-chift-ignored-error-codes"] = ",".join(self.ignored_error_codes)

        if self.extra_context:
            # used to pass extra context about the request to chift
            try:
                headers["x-chift-extra-context"] = base64.b64encode(
                    json.dumps(self.extra_context).encode("utf-8")
                ).decode("utf-8")
            except Exception:
                pass

        try:
            req = self.process_request(
                request_type, url, headers=headers, params=params, json=data
            )
        except requests.exceptions.RetryError as e:
            raise exceptions.ChiftException(
                f"After {self.max_retries} retries, the request failed."
            )
        finally:
            # reset client_request_id and raw_data
            self.client_request_id = None
            self.raw_data = None

        if req.status_code == httplib.UNAUTHORIZED:
            try:
                err = req.json()
            except Exception:
                raise exceptions.ChiftException(
                    "Application authentication failed",
                    error_code=req.status_code,
                    detail=req.text,
                )
            message = err.get("message", "Application authentication failed")
            error_code = err.get("error_code", req.status_code)
            raise exceptions.ChiftException(
                message, error_code=error_code, detail=err.get("detail", req.text)
            )

        if req.status_code == 204:
            return True

        try:
            result = req.json()
        except:
            raise exceptions.ChiftException(f"Error parsing json response: {req.text}")

        if isinstance(result, dict) and result.get("status") == "error":
            raise exceptions.ChiftException(
                result.get("message"),
                error_code=result.get("error_code"),
                detail=result.get("detail"),
            )
        elif req.status_code not in [httplib.OK, httplib.CREATED]:
            raise exceptions.ChiftException(
                f"Error returned with status code '{req.status_code}': {req.text}"
            )
        else:
            return result

    def process_request(
        self, request_type, url_path, headers=None, params=None, json=None
    ):
        engine = self.test_client or self.session
        if self.test_client:
            headers.update(self.test_auth.get_auth_header())
        return engine.request(
            request_type,
            self.url_base + url_path,
            headers=headers,
            params=params,
            json=json,
        )

    def get(self, *args, **kwargs):
        return self.make_request("GET", *args, **kwargs)

    def post(self, *args, **kwargs):
        return self.make_request("POST", *args, **kwargs)

    def patch(self, *args, **kwargs):
        return self.make_request("PATCH", *args, **kwargs)

    def delete(self, *args, **kwargs):
        return self.make_request("DELETE", *args, **kwargs)

    def path_builder(self, ordered_paths):
        if ordered_paths:
            return "/" + "/".join(
                [str(path).strip("/") for path in ordered_paths if path]
            )

    def delete_one(
        self, chift_vertical, chift_model, chift_id, params=None, extra_path=None
    ):
        url_path = self.path_builder(
            [
                "consumers" if self.consumer_id else None,
                self.consumer_id,
                chift_vertical,
                chift_model,
                chift_id,
                extra_path,
            ]
        )

        return self.delete(url_path, params=params)

    def get_one(
        self, chift_vertical, chift_model, chift_id, params=None, extra_path=None
    ):
        url_path = self.path_builder(
            [
                "consumers" if self.consumer_id else None,
                self.consumer_id,
                chift_vertical,
                chift_model,
                chift_id,
                extra_path,
            ]
        )

        return self.get(url_path, params=params)

    def get_all(self, chift_vertical, chift_model, params=None, extra_path=None):
        url_path = self.path_builder(
            [
                "consumers" if self.consumer_id else None,
                self.consumer_id,
                chift_vertical,
                chift_model,
                extra_path,
            ]
        )

        return self.get(url_path, params=params)

    def post_one(self, chift_vertical, chift_model, data, extra_path=None, params=None):
        url_path = self.path_builder(
            [
                "consumers" if self.consumer_id else None,
                self.consumer_id,
                chift_vertical,
                chift_model,
                extra_path,
            ]
        )

        return self.post(url_path, data=data, params=params)

    def update_one(
        self, chift_vertical, chift_model, chift_id, data, extra_path=None, params=None
    ):
        url_path = self.path_builder(
            [
                "consumers" if self.consumer_id else None,
                self.consumer_id,
                chift_vertical,
                chift_model,
                chift_id,
                extra_path,
            ]
        )

        return self.patch(url_path, data=data, params=params)

    def set_ignored_error_codes(self, ignored_error_codes=[]):
        self.ignored_error_codes = ignored_error_codes

    def set_extra_context(self, extra_context=None):
        self.extra_context = extra_context

1	import base64	1✔
2	import http.client as httplib	1✔
3	import json	1✔
4	from datetime import datetime	1✔
5
6	import requests	1✔
7	from requests.adapters import HTTPAdapter	1✔
8	from requests.packages.urllib3.util.retry import Retry	1✔
9
10	from chift.api import exceptions	1✔
11
12
13	class ChiftAuth(requests.auth.AuthBase):	1✔
14	def __init__(	1✔
15	self, client_id, client_secret, account_id, url_base, env_id, test_client
16	):
17	self.client_id = client_id	1✔
18	self.client_secret = client_secret	1✔
19	self.account_id = account_id	1✔
20	self.url_base = url_base	1✔
21	self.env_id = env_id	1✔
22	self.test_client = test_client	1✔
23	self.request_engine = self.test_client or requests	1✔
24
25	self.access_token = None	1✔
26	self.expires_at = None	1✔
27
28	def __call__(self, request):	1✔
29	request.headers.update(self.get_auth_header())	×
30	return request	×
31
32	def get_auth_header(self):	1✔
33	return {"Authorization": f"Bearer {self.get_access_token()}"}	×
34
35	def _parse_token(self, token):	1✔
36	self.access_token = token.get("access_token")	×
37	self.expires_at = datetime.fromtimestamp(token.get("expires_on"))	×
38
39	def get_access_token(self):	1✔
40	if self.access_token:	×
41	if datetime.now() < self.expires_at:	×
42	return self.access_token	×
43
44	payload = {	×
45	"clientId": self.client_id,
46	"clientSecret": self.client_secret,
47	"accountId": self.account_id,
48	}
49	if self.env_id:	×
50	payload["envId"] = self.env_id	×
51
52	response = self.request_engine.post(self.url_base + "/token", json=payload)	×
53
54	if not response.status_code == httplib.OK:	×
55	raise exceptions.ChiftException(	×
56	f"Error while authenticating '{response.status_code}': {response.text}"
57	)
58
59	self._parse_token(response.json())	×
60
61	return self.access_token	×
62
63
64	class ChiftClient:	1✔
65	session = None	1✔
66	access_token = None	1✔
67	auth = None	1✔
68	consumer_id = None	1✔
69	connection_id = None	1✔
70	raw_data = None	1✔
71	client_request_id = None	1✔
72	related_chain_execution_id = None	1✔
73	sync_id = None	1✔
74
75	__instance = None	1✔
76	__use_global = False	1✔
77
78	def __new__(cls, **kwargs):	1✔
79	"""
80	__use_global=True for singleton usage
81	"""
82	if ChiftClient.__use_global:	1✔
83	if ChiftClient.__instance is None:	×
84	ChiftClient.__instance = object.__new__(cls)	×
85	instance = ChiftClient.__instance	×
86	else:
87	instance = object.__new__(cls)	1✔
88
89	return instance	1✔
90
91	def __init__(self, consumer_id=None, **kwargs):	1✔
92	from chift import (	1✔
93	account_id,
94	client_id,
95	client_secret,
96	related_chain_execution_id,
97	sync_id,
98	url_base,
99	)
100
101	self.consumer_id = consumer_id	1✔
102	self.client_id = kwargs.get("client_id") or client_id	1✔
103	self.client_secret = kwargs.get("client_secret") or client_secret	1✔
104	self.account_id = kwargs.get("account_id") or account_id	1✔
105	self.url_base = kwargs.get("url_base") or url_base	1✔
106	self.env_id = kwargs.get("env_id")	1✔
107	self.test_client = kwargs.get("test_client")	1✔
108	self.related_chain_execution_id = (	1✔
109	kwargs.get("related_chain_execution_id") or related_chain_execution_id
110	)
111	self.sync_id = kwargs.get("sync_id") or sync_id	1✔
112	self.max_retries = kwargs.get("max_retries")	1✔
113	self._start_session()	1✔
114
115	def _start_session(self):	1✔
116	self.ignored_error_codes = []	1✔
117	self.extra_context = None	1✔
118	if self.test_client:	1✔
119	self.url_base = "" # set to empty string to avoid url_base in the request	×
120	self.test_auth = ChiftAuth(	×
121	self.client_id,
122	self.client_secret,
123	self.account_id,
124	self.url_base,
125	self.env_id,
126	self.test_client,
127	)
128	elif not self.session:	1✔
129	self.session = requests.Session()	1✔
130	self.session.auth = ChiftAuth(	1✔
131	self.client_id,
132	self.client_secret,
133	self.account_id,
134	self.url_base,
135	self.env_id,
136	None,
137	)
138
139	if self.max_retries:	1✔
140	retries = Retry(	×
141	total=int(self.max_retries),
142	backoff_factor=0.1,
143	status_forcelist=[502],
144	)
145	self.session.mount("http://", HTTPAdapter(max_retries=retries))	×
146	self.session.mount("https://", HTTPAdapter(max_retries=retries))	×
147
148	def make_request(	1✔
149	self, request_type, url, data=None, content_type="application/json", params=None
150	):
151	if not params:	×
152	params = {}	×
153
154	if isinstance(data, str):	×
155	data = json.loads(data)	×
156
157	headers = {	×
158	"Content-Type": content_type,
159	"Accept": "application/json",
160	"User-Agent": "chift-python-sdk library",
161	"Accept-Encoding": "gzip",
162	}
163
164	if self.connection_id:	×
165	headers["x-chift-connectionid"] = self.connection_id	×
166
167	if self.raw_data:	×
168	headers["x-chift-raw-data"] = "true"	×
169
170	if self.client_request_id:	×
171	headers["x-chift-client-requestid"] = self.client_request_id	×
172
173	if self.related_chain_execution_id:	×
174	headers["x-chift-relatedchainexecutionid"] = self.related_chain_execution_id	×
175
176	if self.sync_id:	×
177	headers["x-chift-syncid"] = self.sync_id	×
178
179	if self.ignored_error_codes:	×
180	headers["x-chift-ignored-error-codes"] = ",".join(self.ignored_error_codes)	×
181
182	if self.extra_context:	×
183	# used to pass extra context about the request to chift
184	try:	×
185	headers["x-chift-extra-context"] = base64.b64encode(	×
186	json.dumps(self.extra_context).encode("utf-8")
187	).decode("utf-8")
188	except Exception:	×
189	pass	×
190
191	try:	×
192	req = self.process_request(	×
193	request_type, url, headers=headers, params=params, json=data
194	)
195	except requests.exceptions.RetryError as e:	×
196	raise exceptions.ChiftException(	×
197	f"After {self.max_retries} retries, the request failed."
198	)
199	finally:
200	# reset client_request_id and raw_data
201	self.client_request_id = None	×
202	self.raw_data = None	×
203
204	if req.status_code == httplib.UNAUTHORIZED:	×
NEW 205	try:	×
NEW 206	err = req.json()	×
NEW 207	except Exception:	×
NEW 208	raise exceptions.ChiftException(	×
209	"Application authentication failed",
210	error_code=req.status_code,
211	detail=req.text,
212	)
NEW 213	message = err.get("message", "Application authentication failed")	×
NEW 214	error_code = err.get("error_code", req.status_code)	×
UNCOV 215	raise exceptions.ChiftException(	×
216	message, error_code=error_code, detail=err.get("detail", req.text)
217	)
218
219	if req.status_code == 204:	×
220	return True	×
221
222	try:	×
223	result = req.json()	×
224	except:	×
225	raise exceptions.ChiftException(f"Error parsing json response: {req.text}")	×
226
227	if isinstance(result, dict) and result.get("status") == "error":	×
228	raise exceptions.ChiftException(	×
229	result.get("message"),
230	error_code=result.get("error_code"),
231	detail=result.get("detail"),
232	)
233	elif req.status_code not in [httplib.OK, httplib.CREATED]:	×
234	raise exceptions.ChiftException(	×
235	f"Error returned with status code '{req.status_code}': {req.text}"
236	)
237	else:
238	return result	×
239
240	def process_request(	1✔
241	self, request_type, url_path, headers=None, params=None, json=None
242	):
243	engine = self.test_client or self.session	×
244	if self.test_client:	×
245	headers.update(self.test_auth.get_auth_header())	×
246	return engine.request(	×
247	request_type,
248	self.url_base + url_path,
249	headers=headers,
250	params=params,
251	json=json,
252	)
253
254	def get(self, args, *kwargs):	1✔
255	return self.make_request("GET", args, *kwargs)	1✔
256
257	def post(self, args, *kwargs):	1✔
258	return self.make_request("POST", args, *kwargs)	1✔
259
260	def patch(self, args, *kwargs):	1✔
261	return self.make_request("PATCH", args, *kwargs)	1✔
262
263	def delete(self, args, *kwargs):	1✔
264	return self.make_request("DELETE", args, *kwargs)	×
265
266	def path_builder(self, ordered_paths):	1✔
267	if ordered_paths:	1✔
268	return "/" + "/".join(	1✔
269	[str(path).strip("/") for path in ordered_paths if path]
270	)
271
272	def delete_one(	1✔
273	self, chift_vertical, chift_model, chift_id, params=None, extra_path=None
274	):
275	url_path = self.path_builder(	×
276	[
277	"consumers" if self.consumer_id else None,
278	self.consumer_id,
279	chift_vertical,
280	chift_model,
281	chift_id,
282	extra_path,
283	]
284	)
285
286	return self.delete(url_path, params=params)	×
287
288	def get_one(	1✔
289	self, chift_vertical, chift_model, chift_id, params=None, extra_path=None
290	):
291	url_path = self.path_builder(	1✔
292	[
293	"consumers" if self.consumer_id else None,
294	self.consumer_id,
295	chift_vertical,
296	chift_model,
297	chift_id,
298	extra_path,
299	]
300	)
301
302	return self.get(url_path, params=params)	1✔
303
304	def get_all(self, chift_vertical, chift_model, params=None, extra_path=None):	1✔
305	url_path = self.path_builder(	1✔
306	[
307	"consumers" if self.consumer_id else None,
308	self.consumer_id,
309	chift_vertical,
310	chift_model,
311	extra_path,
312	]
313	)
314
315	return self.get(url_path, params=params)	1✔
316
317	def post_one(self, chift_vertical, chift_model, data, extra_path=None, params=None):	1✔
318	url_path = self.path_builder(	1✔
319	[
320	"consumers" if self.consumer_id else None,
321	self.consumer_id,
322	chift_vertical,
323	chift_model,
324	extra_path,
325	]
326	)
327
328	return self.post(url_path, data=data, params=params)	1✔
329
330	def update_one(	1✔
331	self, chift_vertical, chift_model, chift_id, data, extra_path=None, params=None
332	):
333	url_path = self.path_builder(	1✔
334	[
335	"consumers" if self.consumer_id else None,
336	self.consumer_id,
337	chift_vertical,
338	chift_model,
339	chift_id,
340	extra_path,
341	]
342	)
343
344	return self.patch(url_path, data=data, params=params)	1✔
345
346	def set_ignored_error_codes(self, ignored_error_codes=[]):	1✔
347	self.ignored_error_codes = ignored_error_codes	×
348
349	def set_extra_context(self, extra_context=None):	1✔
350	self.extra_context = extra_context	×

chift-oneapi / chift-python-sdk / 23051079501

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous