71 lines
2.5 KiB
Python
71 lines
2.5 KiB
Python
from base64 import b64encode
|
|
from typing import Union
|
|
|
|
import gssapi
|
|
import requests
|
|
from requests_gssapi import HTTPSPNEGOAuth
|
|
from zope import component
|
|
from zope.interface import implementer
|
|
|
|
from ceo_common.krb5.utils import get_fwd_tgt
|
|
from ceo_common.interfaces import IConfig, IHTTPClient
|
|
|
|
|
|
@implementer(IHTTPClient)
|
|
class HTTPClient:
|
|
def __init__(self):
|
|
# Determine how to connect to other ceod instances
|
|
cfg = component.getUtility(IConfig)
|
|
if cfg.get('ceod_use_https'):
|
|
self.scheme = 'https'
|
|
else:
|
|
self.scheme = 'http'
|
|
self.ceod_port = cfg.get('ceod_port')
|
|
self.base_domain = cfg.get('base_domain')
|
|
|
|
def request(self, host: str, api_path: str, method: str, principal: str,
|
|
need_cred: bool, **kwargs):
|
|
# always use the FQDN
|
|
if '.' not in host:
|
|
host = host + '.' + self.base_domain
|
|
|
|
# SPNEGO
|
|
if principal is not None:
|
|
gssapi_name = gssapi.Name(principal)
|
|
creds = gssapi.Credentials(name=gssapi_name, usage='initiate')
|
|
else:
|
|
creds = None
|
|
auth = HTTPSPNEGOAuth(
|
|
opportunistic_auth=True,
|
|
target_name=gssapi.Name('ceod/' + host),
|
|
creds=creds,
|
|
)
|
|
|
|
# Forwarded TGT (X-KRB5-CRED)
|
|
headers = {}
|
|
if need_cred:
|
|
b = get_fwd_tgt('ceod/' + host)
|
|
headers['X-KRB5-CRED'] = b64encode(b).decode()
|
|
|
|
return requests.request(
|
|
method,
|
|
f'{self.scheme}://{host}:{self.ceod_port}{api_path}',
|
|
auth=auth, headers=headers, **kwargs,
|
|
)
|
|
|
|
def get(self, host: str, api_path: str, principal: Union[str, None] = None,
|
|
need_cred: bool = False, **kwargs):
|
|
return self.request(host, api_path, 'GET', principal, need_cred, **kwargs)
|
|
|
|
def post(self, host: str, api_path: str, principal: Union[str, None] = None,
|
|
need_cred: bool = False, **kwargs):
|
|
return self.request(host, api_path, 'POST', principal, need_cred, **kwargs)
|
|
|
|
def patch(self, host: str, api_path: str, principal: Union[str, None] = None,
|
|
need_cred: bool = False, **kwargs):
|
|
return self.request(host, api_path, 'PATCH', principal, need_cred, **kwargs)
|
|
|
|
def delete(self, host: str, api_path: str, principal: Union[str, None] = None,
|
|
need_cred: bool = False, **kwargs):
|
|
return self.request(host, api_path, 'DELETE', principal, need_cred, **kwargs)
|