FastCF

from fastcore.utils import *
from fastspec.oapi import SpecParser, OpenAPIClient

import httpx

acctok = os.getenv('CLOUDFLARE_ACC_TOK')
usrtok = os.environ['CLOUDFLARE_USR_TOK']
acctid = os.getenv('CLOUDFLARE_ACCT_ID')
acceml = os.getenv('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

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

glbtok = os.getenv('CLOUDFLARE_API_KEY')
if acceml and glbtok:
    ghdrs = {'X-Auth-Email': acceml, 'X-Auth-Key': glbtok, 'Content-Type': 'application/json'}
    r = httpx.get(f'{base}/user', headers=ghdrs)
    print(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')]

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

CloudflareApi

_filler = {'for','a','the','an','in','of'}

def _tag2res(tag):
    parts = tag.lower().replace('-','_').replace('(','').replace(')','').split()
    return '_'.join(p for p in parts if p not in _filler)

def _op2name(op_id, tag, path='', verb=''):
    nm = camel2snake(op_id).replace('-', '_')
    nm = re.sub(r'_0_', '_', nm).strip('_')
    res_parts = _tag2res(tag).split('_')
    nm_parts = nm.split('_')
    while res_parts and nm_parts:
        if nm_parts[0] in _filler: nm_parts.pop(0); continue
        if nm_parts[0].rstrip('s') == res_parts[0].rstrip('s'): res_parts.pop(0); nm_parts.pop(0)
        else: break
    nm = '_'.join(nm_parts)
    if not nm:
        seg = path.rstrip('/').rsplit('/', 1)[-1].replace('-','_').strip('{}')
        nm = f'{verb}_{seg}'
    return nm

def cf_group_func(op_id, path, verb='', path_tags=None, op_tags=None):
    tag = first(op_tags or path_tags) or 'misc'
    return _tag2res(tag), _op2name(op_id, tag, path, verb)

def _cf_spec():
    _pkg = Path(__file__).parent if '__file__' in globals() else Path('../fastcflare')
    schema_dict = dict2obj((_pkg/'openapi.json').read_json())
    return SpecParser.from_openapi(schema_dict, cf_group_func)

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']

class CloudflareApi:
    "Cloudflare API client supporting both user and account tokens"
    def __init__(self, token=None, account_id=None, email=None, api_key=None):
        if not token and not api_key: token = os.getenv('CF_API_TOKEN')
        if api_key and not email: email = os.getenv('CF_API_EMAIL')
        store_attr()
        if api_key: hdrs = {'X-Auth-Email': email, 'X-Auth-Key': api_key, 'Content-Type': 'application/json'}
        else: hdrs = {'Authorization': f'Bearer {token}', 'Content-Type': 'application/json'}
        self.cli = OpenAPIClient(_cf_spec(), headers=hdrs, timeout=60.0)
        self.groups = self.cli.groups

    async def verify(self):
        if self.account_id: return await self.account_owned_api_tokens.api_tokens_verify_token(account_id=self.account_id)
        return await self.user_api_tokens.verify_token()

    def __dir__(self): return super().__dir__() + list(self.groups)
    def _repr_markdown_(self): return "\n".join(f"- {o}" for o in sorted(self.groups))
    def __getattr__(self, k): return self.groups[k] if 'groups' in vars(self) and k in self.groups else stop(AttributeError(k))

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

'active'

if acctok and acctid:
    acf = CloudflareApi(token=acctok, account_id=acctid)
    print((await 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'

if acctok and acctid:
    print((await acf.dns_records_zone.list_dns_records(zone_id=zid)).result[0].name)

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

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 dict2obj(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[38], 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]

File ~/aai-ws/fastspec/fastspec/oapi.py:193, in __call__(self, *args, **kwargs)
    191 else: kw = dict(body=body)
    192 if stream: return self._stream(url, headers=headers, query=query, route=route, **kw)
--> 193 return 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 dict2obj(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 dict2obj(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[39], line 1
----> 1 tok = await ucf.create_token(['answer.ai'], ['Zone Read'], 'fastcflare-test-token')
      2 tok.success, tok.result.name

Cell In[37], 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)])

File ~/aai-ws/fastspec/fastspec/oapi.py:193, in __call__(self, *args, **kwargs)
    191 else: kw = dict(body=body)
    192 if stream: return self._stream(url, headers=headers, query=query, route=route, **kw)
--> 193 return 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 dict2obj(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)

Clean up the token we just created:

res = await ucf.user_api_tokens.delete_token(tok.result.id)
assert res.success