fastcflare

FastCF

import httpx

# acctok = os.environ['CLOUDFLARE_ACC_TOK']
usrtok = os.environ['CLOUDFLARE_USR_TOK']
# acctid = os.environ['CLOUDFLARE_ACCT_ID']
# acceml = os.environ['CLOUDFLARE_EML_ADD']

base = 'https://api.cloudflare.com/client/v4'

uhdrs = {'Authorization': f'Bearer {usrtok}'}
r = httpx.get(f'{base}/user/tokens/verify', headers=uhdrs)
r.json()['success']

True

# ahdrs = {'Authorization': f'Bearer {acctok}'}
# r = httpx.get(f'{base}/accounts/{acctid}/tokens/verify', headers=ahdrs)
# r.json()['success']

# glbtok = os.environ['CLOUDFLARE_API_KEY']
# 
# ghdrs = {'X-Auth-Email': acceml, 'X-Auth-Key': glbtok, 'Content-Type': 'application/json'}
# r = httpx.get(f'{base}/user', headers=ghdrs)
# r.json()['success']

r = httpx.get(f'{base}/zones', headers=uhdrs)
[(z['name'], z['id'][:8]) for z in r.json()['result'][:1]]

[('answer.ai', 'a00d788b')]

# r = httpx.get(f'{base}/zones', headers=ahdrs, params={'account.id': acctid})
# [(z['name'], z['id'][:8]) for z in r.json()['result'][:1]]

CloudflareApi

source

cf_group_func


def cf_group_func(
    op_id, path, verb:str='', path_tags:NoneType=None, op_tags:NoneType=None
):

Call self as a function.

cli = OpenAPIClient(_cf_spec(), headers={'Authorization': f'Bearer {usrtok}', 'Content-Type': 'application/json'})
list(cli.groups)[:10]

['accounts',
 'applications',
 'category',
 'custom_pages_account',
 'tseng_abuse_complaint_processor_other',
 'mcp_portal',
 'mcp_portal_servers',
 'access_applications',
 'access_short_lived_certificate_cas',
 'access_application_scoped_policies']

source

CloudflareApi


def CloudflareApi(
    token:NoneType=None, account_id:NoneType=None, email:NoneType=None, api_key:NoneType=None
):

Cloudflare API client supporting both user and account tokens

patch OpFunc.__call__ to return dict2obj

source

OpFunc.call


async def __call__(
    args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

Call self as a function.

ucf = CloudflareApi(token=usrtok)
(await ucf.verify()).result.status

'active'

# acf = CloudflareApi(token=acctok, account_id=acctid)
# acf.verify().result.status

zone_id is parsed as a required param, so we pass an empty string:

ucf.zone.get

Zone Details

Parameters: - zone_id (str, required)

(await ucf.zone.get('')).result[0].name

'answer.ai'

zid = (await ucf.zone.get('')).result[0].id
(await ucf.dns_records_zone.list_dns_records(zone_id=zid)).result[0].name

'compose.answer.ai'

# (await acf.dns_records_zone.list_dns_records(zone_id=zid)).result[0].name

source

CloudflareApi.create_token


async def create_token(
    doms, perm_names, name, grp:str='account.zone.'
):

Create a scoped Cloudflare API token for given domains and permission names

pgs = (await ucf.user_api_tokens.permission_groups_list_permission_groups()).result
[p.name for p in pgs if 'DNS' in p.name or 'Zone' in p.name][:20]

---------------------------------------------------------------------------
HTTPStatusError                           Traceback (most recent call last)
File ~/aai-ws/fastspec/fastspec/oapi.py:171, in _request(self, url, headers, query, body, route, **kwargs)
    170 "Execute an HTTP request and return decoded response."
--> 171 try: return await self.client.request(self.verb, url, headers=headers, params=query, json_data=body, **kwargs)
    172 except Exception as e: self._raise_with_context(e, endpoint='', route=route, query=query, body=body)

File ~/aai-ws/fastspec/fastspec/transport.py:52, in AsyncTransport.request(self, method, url, headers, params, json_data, data, files, raw)
     50 resp = await self.client.request(method, url, headers=self._request_headers(headers, files=files),
     51     params=params, json=json_data, data=data, files=files)
---> 52 try: resp.raise_for_status()
     53 except httpx.HTTPStatusError as e:

File ~/aai-ws/.venv/lib/python3.12/site-packages/httpx/_models.py:829, in Response.raise_for_status(self)
    828 message = message.format(self, error_type=error_type)
--> 829 raise HTTPStatusError(message, request=request, response=self)

HTTPStatusError: Client error '403 Forbidden' for url 'https://api.cloudflare.com/client/v4/user/tokens/permission_groups'
For more information check: https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/403
{"success":false,"errors":[{"code":9109,"message":"Unauthorized to access requested resource"}],"messages":[],"result":null}

The above exception was the direct cause of the following exception:

APIError                                  Traceback (most recent call last)
Cell In[70], line 1
----> 1 pgs = (await ucf.user_api_tokens.permission_groups_list_permission_groups()).result
      2 [p.name for p in pgs if 'DNS' in p.name or 'Zone' in p.name][:20]

Cell In[62], line 9, in OpFunc.__call__(self, *args, **kwargs)
      5     if files: kw = dict(body=None, files=files, data=self.form_encoder(body) or None)
      6     elif self.request_content_type == "application/x-www-form-urlencoded": kw = dict(body=None, data=self.form_encoder(body))
      7     else: kw = dict(body=body)
      8     if stream: return self._stream(url, headers=headers, query=query, route=route, **kw)
----> 9     return dict2obj(await self._request(url, headers=headers, query=query, route=route, **kw))

File ~/aai-ws/fastspec/fastspec/oapi.py:172, in _request(self, url, headers, query, body, route, **kwargs)
    170 "Execute an HTTP request and return decoded response."
    171 try: return await self.client.request(self.verb, url, headers=headers, params=query, json_data=body, **kwargs)
--> 172 except Exception as e: self._raise_with_context(e, endpoint='', route=route, query=query, body=body)

File ~/aai-ws/fastspec/fastspec/oapi.py:162, in _raise_with_context(self, exc, endpoint, route, query, body)
    160 # TODO: Make APIError generic, users can modify/subclass it include additional info like model,provider etc..
    161 if isinstance(exc, (httpx.HTTPStatusError, httpx.RequestError)):
--> 162     raise exc.api_error(provider=provider, model=model) from exc
    163 raise exc

APIError: APIError(message='{"success": false, "errors": [{"code": 9109, "message": "Unauthorized to access requested resource"}], "messages": [], "result": null}', endpoint='GET /client/v4/user/tokens/permission_groups', status_code=403)

tok = await ucf.create_token(['answer.ai'], ['Zone Read'], 'fastcflare-test-token')
tok.success, tok.result.name

---------------------------------------------------------------------------
HTTPStatusError                           Traceback (most recent call last)
File ~/aai-ws/fastspec/fastspec/oapi.py:171, in _request(self, url, headers, query, body, route, **kwargs)
    170 "Execute an HTTP request and return decoded response."
--> 171 try: return await self.client.request(self.verb, url, headers=headers, params=query, json_data=body, **kwargs)
    172 except Exception as e: self._raise_with_context(e, endpoint='', route=route, query=query, body=body)

File ~/aai-ws/fastspec/fastspec/transport.py:52, in AsyncTransport.request(self, method, url, headers, params, json_data, data, files, raw)
     50 resp = await self.client.request(method, url, headers=self._request_headers(headers, files=files),
     51     params=params, json=json_data, data=data, files=files)
---> 52 try: resp.raise_for_status()
     53 except httpx.HTTPStatusError as e:

File ~/aai-ws/.venv/lib/python3.12/site-packages/httpx/_models.py:829, in Response.raise_for_status(self)
    828 message = message.format(self, error_type=error_type)
--> 829 raise HTTPStatusError(message, request=request, response=self)

HTTPStatusError: Client error '403 Forbidden' for url 'https://api.cloudflare.com/client/v4/user/tokens/permission_groups'
For more information check: https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/403
{"success":false,"errors":[{"code":9109,"message":"Unauthorized to access requested resource"}],"messages":[],"result":null}

The above exception was the direct cause of the following exception:

APIError                                  Traceback (most recent call last)
Cell In[71], line 1
----> 1 tok = await ucf.create_token(['answer.ai'], ['Zone Read'], 'fastcflare-test-token')
      2 tok.success, tok.result.name

Cell In[69], line 5, in CloudflareApi.create_token(self, doms, perm_names, name, grp)
      1 @patch
      2 async def create_token(self:CloudflareApi, doms, perm_names, name, grp='account.zone.'):
      3     "Create a scoped Cloudflare API token for given domains and permission names"
      4     pref = 'com.cloudflare.api.'+grp
----> 5     pgs = (await self.user_api_tokens.permission_groups_list_permission_groups()).result
      6     perms = [dict(id=p.id) for p in pgs if p.name in perm_names]
      7     zids = {f'{pref}{(await self.zone.get(name=d)).result[0].id}':'*' for d in doms}
      8     return await self.user_api_tokens.create_token(name=name, policies=[dict(effect='allow', resources=zids, permission_groups=perms)])

Cell In[62], line 9, in OpFunc.__call__(self, *args, **kwargs)
      5     if files: kw = dict(body=None, files=files, data=self.form_encoder(body) or None)
      6     elif self.request_content_type == "application/x-www-form-urlencoded": kw = dict(body=None, data=self.form_encoder(body))
      7     else: kw = dict(body=body)
      8     if stream: return self._stream(url, headers=headers, query=query, route=route, **kw)
----> 9     return dict2obj(await self._request(url, headers=headers, query=query, route=route, **kw))

File ~/aai-ws/fastspec/fastspec/oapi.py:172, in _request(self, url, headers, query, body, route, **kwargs)
    170 "Execute an HTTP request and return decoded response."
    171 try: return await self.client.request(self.verb, url, headers=headers, params=query, json_data=body, **kwargs)
--> 172 except Exception as e: self._raise_with_context(e, endpoint='', route=route, query=query, body=body)

File ~/aai-ws/fastspec/fastspec/oapi.py:162, in _raise_with_context(self, exc, endpoint, route, query, body)
    160 # TODO: Make APIError generic, users can modify/subclass it include additional info like model,provider etc..
    161 if isinstance(exc, (httpx.HTTPStatusError, httpx.RequestError)):
--> 162     raise exc.api_error(provider=provider, model=model) from exc
    163 raise exc

APIError: APIError(message='{"success": false, "errors": [{"code": 9109, "message": "Unauthorized to access requested resource"}], "messages": [], "result": null}', endpoint='GET /client/v4/user/tokens/permission_groups', status_code=403)