public
/
pyceo
27
1
Fork 3
Code Issues 25 Pull Requests 5 Projects Releases 1 Wiki Activity

add Kerberos delegation

This commit is contained in:
Max Erenberg 2021-08-18 01:59:24 +00:00
parent d82b5a763b
commit dd59bea918
36 changed files with 853 additions and 174 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.gitignore vendored
View File

@ -1,3 +1,7 @@
__pycache__/
/venv/
.vscode/
/cred
*.o
*.so
/ceo_common/krb5/_krb5.c

34
README.md
View File

@ -9,13 +9,25 @@ this repo in one of the dev environment containers.
Next, install and activate a virtualenv:
```sh
sudo apt install libkrb5-dev python3-dev
python3 -m venv venv
. venv/bin/activate
pip install -r requirements.txt
pip install -r dev-requirements.txt
```
### Running the application
## C bindings
Due to the lack of a decent Python library for Kerberos we ended up
writing our own C bindings using [cffi](https://cffi.readthedocs.io).
Make sure you compile the bindings:
```sh
cd ceo_common/krb5
python krb5_build.py
```
This should create a file named '_krb5.cpython-37m-x86_64-linux-gnu.so'.
This will be imported by other modules in ceo.
## Running the application
ceod is essentially a distributed application, with instances on different
hosts offering different services. For example, the ceod instance on mail
offers a service to subscribe people to mailing lists, and
@ -35,13 +47,14 @@ Sometimes changes you make in the source code don't show up while Flask
is running. Stop the flask app (Ctrl-C), run `clear_cache.sh`, then
restart the app.
### Interacting with the application
## Interacting with the application
The client part of ceo hasn't been written yet, so we'll use curl to
interact with ceod for now.
ceod uses [SPNEGO](https://en.wikipedia.org/wiki/SPNEGO) for authentication,
and TLS for confidentiality and integrity. In development mode, TLS can be
disabled.
First, make sure that your version of curl has been compiled with SPNEGO
support:
```sh
@ -49,9 +62,22 @@ curl -V
```
Your should see 'SPNEGO' in the 'Features' section.
Here's an example of using curl with SPNEGO:
The API also uses unconstrained Kerberos delegation when interacting with
the LDAP database. This means that the client obtains a forwarded TGT, then
sends that to ceod, which then uses it to interact with LDAP on the client's
behalf. There is a script called `gen_cred.py` which can generate this
ticket for you.
Here's an example of making a request to an endpoint which writes to LDAP:
```sh
# Get a Kerberos TGT first
kinit
curl --negotiate -u : --service-name ceod -X POST http://mail:9987/api/mailman/csc-general/ctdalek
# Obtain a forwarded TGT
./gen_cred.py phosphoric-acid
# Make the request
curl --negotiate -u : --service-name ceod \
-H "X-KRB5-CRED: $(cat cred)" \
-d '{"uid":"test_1","cn":"Test One","program":"Math","terms":["s2021"]}' \
-X POST http://phosphoric-acid:9987/api/members
```

4
ceo_common/errors.py
View File

@ -12,6 +12,10 @@ class BadRequest(Exception):
pass
class KerberosError(Exception):
pass
class UserAlreadyExistsError(Exception):
def __init__(self):
super().__init__('user already exists')

3
ceo_common/interfaces/IKerberosService.py
View File

@ -4,9 +4,6 @@ from zope.interface import Interface
class IKerberosService(Interface):
"""A utility wrapper around kinit/kadmin."""
def kinit():
"""Acquire and cache a new TGT."""
def addprinc(principal: str, password: str):
"""Add a new principal with the specified password."""

0
ceo_common/krb5/__init__.py Normal file
View File

96
ceo_common/krb5/krb5_build.py Normal file
View File

@ -0,0 +1,96 @@
from cffi import FFI
ffibuilder = FFI()
# Definitions selectively copied from <krb5/krb5.h>.
# Add more if necessary.
ffibuilder.cdef(r"""
# define KV5M_DATA ...
typedef int32_t krb5_int32;
typedef krb5_int32 krb5_error_code;
typedef krb5_error_code krb5_magic;
struct _krb5_context;
typedef struct _krb5_context* krb5_context;
struct _krb5_auth_context;
typedef struct _krb5_auth_context * krb5_auth_context;
struct _krb5_ccache;
typedef struct _krb5_ccache *krb5_ccache;
typedef struct krb5_principal_data {
...;
} krb5_principal_data;
typedef krb5_principal_data * krb5_principal;
typedef const krb5_principal_data *krb5_const_principal;
typedef struct _krb5_creds {
krb5_principal client;
krb5_principal server;
...;
} krb5_creds;
typedef struct _krb5_data {
krb5_magic magic;
unsigned int length;
char *data;
} krb5_data;
typedef struct krb5_replay_data {
...;
} krb5_replay_data;
krb5_error_code krb5_init_context(krb5_context * context);
void krb5_free_context(krb5_context context);
krb5_error_code krb5_auth_con_init(
krb5_context context, krb5_auth_context *auth_context);
krb5_error_code krb5_auth_con_setflags(
krb5_context context, krb5_auth_context auth_context, krb5_int32 flags);
krb5_error_code krb5_auth_con_free(
krb5_context context, krb5_auth_context auth_context);
krb5_error_code
krb5_cc_new_unique(krb5_context context, const char *type, const char *hint,
krb5_ccache *id);
krb5_error_code krb5_cc_default(krb5_context context, krb5_ccache *ccache);
krb5_error_code krb5_cc_resolve(
krb5_context context, const char * name, krb5_ccache * cache);
krb5_error_code
krb5_cc_initialize(krb5_context context, krb5_ccache cache,
krb5_principal principal);
krb5_error_code
krb5_cc_get_principal(krb5_context context, krb5_ccache cache,
krb5_principal *principal);
krb5_error_code krb5_cc_store_cred(
krb5_context context, krb5_ccache cache, krb5_creds *creds);
krb5_error_code krb5_cc_close(krb5_context context, krb5_ccache cache);
krb5_error_code krb5_cc_destroy(krb5_context context, krb5_ccache cache);
krb5_error_code
krb5_build_principal(krb5_context context,
krb5_principal * princ,
unsigned int rlen,
const char * realm, ...);
krb5_error_code
krb5_parse_name(krb5_context context, const char *name,
krb5_principal *principal_out);
void krb5_free_principal(krb5_context context, krb5_principal val);
krb5_error_code
krb5_rd_cred(krb5_context context, krb5_auth_context auth_context,
krb5_data *pcreddata, krb5_creds ***pppcreds,
krb5_replay_data *outdata);
krb5_error_code krb5_fwd_tgt_creds(
krb5_context context, krb5_auth_context auth_context,
const char * rhost, krb5_principal client, krb5_principal server,
krb5_ccache cc, int forwardable, krb5_data * outbuf);
void krb5_free_tgt_creds(krb5_context context, krb5_creds ** tgts);
void krb5_free_data_contents(krb5_context context, krb5_data * val);
krb5_error_code krb5_unparse_name(
krb5_context context, krb5_const_principal principal, char **name);
void krb5_free_unparsed_name(krb5_context context, char *val);
const char * krb5_get_error_message(krb5_context ctx, krb5_error_code code);
void krb5_free_error_message(krb5_context ctx, const char * msg);
""")
ffibuilder.set_source(
"_krb5",
"""
#include <krb5/krb5.h>
""",
libraries=['krb5']
)
if __name__ == '__main__':
ffibuilder.compile(verbose=True)

201
ceo_common/krb5/utils.py Normal file
View File

@ -0,0 +1,201 @@
import contextlib
import functools
from typing import Union
from ._krb5 import ffi, lib
from ceo_common.errors import KerberosError
def check_rc(k_ctx, rc: int):
"""
Check the return code of a krb5 function.
An exception is raised if rc != 0.
"""
if rc == 0:
return
c_msg = lib.krb5_get_error_message(k_ctx, rc)
msg = ffi.string(c_msg).decode()
lib.krb5_free_error_message(k_ctx, c_msg)
raise KerberosError(msg)
@contextlib.contextmanager
def get_krb5_context():
"""Yields a krb5_context."""
k_ctx = None
try:
p_k_ctx = ffi.new('krb5_context *')
rc = lib.krb5_init_context(p_k_ctx)
k_ctx = p_k_ctx[0] # dereference the pointer
check_rc(k_ctx, rc)
yield k_ctx
finally:
if k_ctx is not None:
lib.krb5_free_context(k_ctx)
@contextlib.contextmanager
def get_krb5_auth_context(k_ctx):
"""Yields a krb5_auth_context."""
a_ctx = None
try:
p_a_ctx = ffi.new('krb5_auth_context *')
rc = lib.krb5_auth_con_init(k_ctx, p_a_ctx)
a_ctx = p_a_ctx[0]
check_rc(k_ctx, rc)
# clear the flags which enable the replay cache
rc = lib.krb5_auth_con_setflags(k_ctx, a_ctx, 0)
check_rc(k_ctx, rc)
yield a_ctx
finally:
if a_ctx is not None:
lib.krb5_auth_con_free(k_ctx, a_ctx)
@contextlib.contextmanager
def get_krb5_cc_default(k_ctx):
"""Yields the default krb5_ccache."""
cache = None
try:
p_cache = ffi.new('krb5_ccache *')
rc = lib.krb5_cc_default(k_ctx, p_cache)
check_rc(k_ctx, rc)
cache = p_cache[0]
yield cache
finally:
if cache is not None:
lib.krb5_cc_close(k_ctx, cache)
@contextlib.contextmanager
def get_krb5_cc_resolve(k_ctx, name: str):
"""
Resolve a credential cache name.
`name` should have the format 'type:residual'.
"""
cache = None
try:
c_name = ffi.new('char[]', name.encode())
p_cache = ffi.new('krb5_ccache *')
rc = lib.krb5_cc_resolve(k_ctx, c_name, p_cache)
check_rc(k_ctx, rc)
cache = p_cache[0]
yield cache
finally:
if cache is not None:
lib.krb5_cc_close(k_ctx, cache)
def get_fwd_tgt(server: str, cache_name: Union[str, None] = None) -> bytes:
"""
Get a forwarded TGT formatted as a KRB-CRED message.
`server` should have the format 'service/host',
e.g. 'ceod/phosphoric-acid.csclub.uwaterloo.ca'.
If `cache_name` is None, the default cache will be used; otherwise,
the cache with that name will be used.
"""
if cache_name is None:
cache_function = get_krb5_cc_default
else:
cache_function = functools.partial(get_krb5_cc_resolve, name=cache_name)
with get_krb5_context() as k_ctx, \
get_krb5_auth_context(k_ctx) as a_ctx, \
cache_function(k_ctx) as cache:
server_princ = None
client_princ = None
outbuf = None
try:
# create a server principal from the server string
c_server = ffi.new('char[]', server.encode())
p_server_princ = ffi.new('krb5_principal *')
rc = lib.krb5_parse_name(k_ctx, c_server, p_server_princ)
check_rc(k_ctx, rc)
server_princ = p_server_princ[0]
# get a client principal from the default cache
p_client_princ = ffi.new('krb5_principal *')
rc = lib.krb5_cc_get_principal(k_ctx, cache, p_client_princ)
check_rc(k_ctx, rc)
client_princ = p_client_princ[0]
# get the forwarded TGT
p_outbuf = ffi.new('krb5_data *')
rc = lib.krb5_fwd_tgt_creds(
k_ctx, a_ctx, ffi.NULL, client_princ, server_princ,
cache, 0, p_outbuf)
check_rc(k_ctx, rc)
outbuf = p_outbuf[0]
return ffi.unpack(outbuf.data, outbuf.length)
finally:
if outbuf is not None:
lib.krb5_free_data_contents(k_ctx, p_outbuf)
if client_princ is not None:
lib.krb5_free_principal(k_ctx, client_princ)
if server_princ is not None:
lib.krb5_free_principal(k_ctx, server_princ)
@contextlib.contextmanager
def store_fwd_tgt_creds(cred_data_bytes: bytes):
"""
Stores the credentials found in cred_data_bytes
in a new sub-cache in the default cache (which should be of type DIR),
yields the name of the principal in the credential, then finally removes
the sub-cache from the collection.
The principal name will have the form 'username@REALM', e.g.
'ctdalek@CSCLUB.UWATERLOO.CA'.
"""
with get_krb5_context() as k_ctx, get_krb5_auth_context(k_ctx) as a_ctx:
creds_out = None
cache = None
c_name = None
try:
# fill a krb5_data struct
cred_data = ffi.new('krb5_data *')
cred_data_buf = ffi.new('char[]', cred_data_bytes)
cred_data.magic = lib.KV5M_DATA
cred_data.length = len(cred_data_bytes)
cred_data.data = cred_data_buf
# read the KRB5-CRED message into a krb5_creds array
p_creds_out = ffi.new('krb5_creds ***')
rc = lib.krb5_rd_cred(k_ctx, a_ctx, cred_data, p_creds_out, ffi.NULL)
check_rc(k_ctx, rc)
creds_out = p_creds_out[0]
# there should only be one cred in the array
cred = creds_out[0]
# read the name of the client principal
client_princ = cred.client
p_name = ffi.new('char **')
rc = lib.krb5_unparse_name(k_ctx, client_princ, p_name)
check_rc(k_ctx, rc)
c_name = p_name[0]
name = ffi.string(c_name).decode()
# create a new cache inside the collection (default cache)
p_cache = ffi.new('krb5_ccache *')
cctype = ffi.new('char[]', b'DIR')
rc = lib.krb5_cc_new_unique(k_ctx, cctype, ffi.NULL, p_cache)
check_rc(k_ctx, rc)
cache = p_cache[0]
# initialize the new cache
rc = lib.krb5_cc_initialize(k_ctx, cache, client_princ)
check_rc(k_ctx, rc)
# store the cred into the cache
rc = lib.krb5_cc_store_cred(k_ctx, cache, cred)
check_rc(k_ctx, rc)
yield name
finally:
if cache is not None:
# We destroy the cache (instead of closing it) since we want
# to remove it from disk.
lib.krb5_cc_destroy(k_ctx, cache)
if c_name is not None:
lib.krb5_free_unparsed_name(k_ctx, c_name)
if creds_out is not None:
lib.krb5_free_tgt_creds(k_ctx, creds_out)

2
ceo_common/model/Config.py
View File

@ -27,4 +27,6 @@ class Config:
return True
if val.lower() in ['false', 'no']:
return False
if section.startswith('auxiliary '):
return val.split(',')
return val

32
ceo_common/model/HTTPClient.py
View File

@ -1,5 +1,6 @@
import socket
from flask import g
import gssapi
from gssapi.raw.exceptions import ExpiredCredentialsError
import requests
@ -22,35 +23,18 @@ class HTTPClient:
self.ceod_port = cfg.get('ceod_port')
self.base_domain = cfg.get('base_domain')
# Determine which principal to use for SPNEGO
# TODO: this code is duplicated in app_factory.py. Figure out
# how to write it only once.
if socket.gethostname() == cfg.get('ceod_admin_host'):
spnego_principal = cfg.get('ldap_admin_principal')
else:
spnego_principal = f'ceod/{socket.getfqdn()}'
# Initialize variables to get Kerberos cache tickets
krb_realm = cfg.get('ldap_sasl_realm')
self.gssapi_name = gssapi.Name(f'{spnego_principal}@{krb_realm}')
self.krb_srv = component.getUtility(IKerberosService)
def get_creds(self):
"""Get GSSAPI credentials to use for SPNEGO."""
for _ in range(2):
try:
creds = gssapi.Credentials(name=self.gssapi_name, usage='initiate')
creds.inquire()
return creds
except ExpiredCredentialsError:
self.krb_srv.kinit()
raise Exception('could not acquire GSSAPI credentials')
self.krb_realm = cfg.get('ldap_sasl_realm')
def request(self, host: str, api_path: str, method='GET', **kwargs):
principal = g.sasl_user
if '@' not in principal:
principal = principal + '@' + self.krb_realm
gssapi_name = gssapi.Name(principal)
creds = gssapi.Credentials(name=gssapi_name, usage='initiate')
auth = HTTPSPNEGOAuth(
opportunistic_auth=True,
target_name='ceod',
creds=self.get_creds(),
creds=creds,
)
# always use the FQDN, for HTTPS purposes
if '.' not in host:

3
ceod/api/app_factory.py
View File

@ -7,6 +7,7 @@ from flask_kerberos import init_kerberos
from zope import component
from .error_handlers import register_error_handlers
from .krb5_cred_handlers import before_request, teardown_request
from ceo_common.interfaces import IConfig, IKerberosService, ILDAPService, IFileService, \
IMailmanService, IMailService, IUWLDAPService, IHTTPClient
from ceo_common.model import Config, HTTPClient, RemoteMailmanService
@ -42,6 +43,8 @@ def create_app(flask_config={}):
app.register_blueprint(uwldap.bp, url_prefix='/api/uwldap')
register_error_handlers(app)
app.before_request(before_request)
app.teardown_request(teardown_request)
@app.route('/ping')
def ping():

29
ceod/api/krb5_cred_handlers.py Normal file
View File

@ -0,0 +1,29 @@
from base64 import b64decode
import traceback
from flask import g, request
from ceo_common.logger_factory import logger_factory
from ceo_common.krb5.utils import store_fwd_tgt_creds
logger = logger_factory(__name__)
def before_request():
if 'x-krb5-cred' not in request.headers:
return
cred = b64decode(request.headers['x-krb5-cred'])
ctx = store_fwd_tgt_creds(cred)
name = ctx.__enter__()
g.stored_creds_ctx = ctx
g.sasl_user = name
def teardown_request(err):
if 'stored_creds_ctx' not in g:
return
try:
ctx = g.stored_creds_ctx
ctx.__exit__(None, None, None)
except Exception:
logger.error(traceback.format_exc())

20
ceod/api/members.py
View File

@ -3,15 +3,16 @@ from zope import component
from .utils import authz_restrict_to_staff, authz_restrict_to_syscom, \
user_is_in_group, requires_authentication_no_realm, \
create_streaming_response, create_sync_response
create_streaming_response, create_sync_response, development_only
from ceo_common.errors import UserNotFoundError
from ceo_common.interfaces import ILDAPService
from ceod.transactions.members import (
AddMemberTransaction,
ModifyMemberTransaction,
RenewMemberTransaction,
ResetPasswordTransaction,
DeleteMemberTransaction,
)
import ceod.utils as utils
bp = Blueprint('members', __name__)
@ -78,5 +79,16 @@ def renew_user(username: str):
@bp.route('/<username>/pwreset', methods=['POST'])
@authz_restrict_to_syscom
def reset_user_password(username: str):
txn = ResetPasswordTransaction(username)
return create_sync_response(txn)
user = component.getUtility(ILDAPService).get_user(username)
password = utils.gen_password()
user.change_password(password)
return {'password': password}
@bp.route('/<username>', methods=['DELETE'])
@authz_restrict_to_syscom
@development_only
def delete_user(username: str):
txn = DeleteMemberTransaction(username)
return create_streaming_response(txn)

16
ceod/api/utils.py
View File

@ -6,7 +6,7 @@ import pwd
import traceback
from typing import Callable, List
from flask import current_app
from flask import current_app, stream_with_context
from flask.json import jsonify
from flask_kerberos import requires_authentication
@ -103,7 +103,8 @@ def create_streaming_response(txn: AbstractTransaction):
'error': str(err),
}) + '\n'
return current_app.response_class(generate(), mimetype='text/plain')
return current_app.response_class(
stream_with_context(generate()), mimetype='text/plain')
def create_sync_response(txn: AbstractTransaction):
@ -123,3 +124,14 @@ def create_sync_response(txn: AbstractTransaction):
return {
'error': str(err),
}, 500
def development_only(f: Callable) -> Callable:
@functools.wraps(f)
def wrapper(*args, **kwargs):
if current_app.config.get('ENV') == 'development':
return f(*args, **kwargs)
return {
'error': 'This endpoint may only be called in development'
}, 403
return wrapper

9
ceod/api/uwldap.py
View File

@ -2,9 +2,8 @@ from flask import Blueprint, request
from flask.json import jsonify
from zope import component
from .utils import create_sync_response, authz_restrict_to_syscom
from .utils import authz_restrict_to_syscom
from ceo_common.interfaces import IUWLDAPService, ILDAPService
from ceod.transactions.uwldap import UpdateProgramsTransaction
bp = Blueprint('uwldap', __name__)
@ -26,9 +25,9 @@ def update_programs():
ldap_srv = component.getUtility(ILDAPService)
body = request.get_json(force=True)
members = body.get('members')
kwargs = {'members': members}
if body.get('dry_run'):
members['dry_run'] = True
return jsonify(
ldap_srv.update_programs(dry_run=True, members=members)
ldap_srv.update_programs(**kwargs)
)
txn = UpdateProgramsTransaction(members=members)
return create_sync_response(txn)

45
ceod/model/KerberosService.py
View File

@ -1,9 +1,14 @@
import os
import shutil
import subprocess
from zope import component
from zope.interface import implementer
from ceo_common.interfaces import IKerberosService
from ceo_common.interfaces import IKerberosService, IConfig
from ceo_common.krb5._krb5 import ffi, lib
from ceo_common.krb5.utils import check_rc, get_krb5_context, \
get_krb5_cc_default
@implementer(IKerberosService)
@ -11,15 +16,39 @@ class KerberosService:
def __init__(
self,
admin_principal: str,
cache_dir: str = '/run/ceod/krb5_cache',
):
self.admin_principal = admin_principal
os.makedirs(cache_dir, exist_ok=True)
os.environ['KRB5CCNAME'] = 'DIR:' + cache_dir
self.kinit()
cfg = component.getUtility(IConfig)
def kinit(self):
subprocess.run(['kinit', '-k', self.admin_principal], check=True)
self.admin_principal = admin_principal
self.cache_dir = cfg.get('ceod_krb5_cache_dir')
self.realm = cfg.get('ldap_sasl_realm')
self._initialize_cache()
def _initialize_cache(self, **kwargs):
if os.path.isdir(self.cache_dir):
shutil.rmtree(self.cache_dir)
os.makedirs(self.cache_dir)
os.environ['KRB5CCNAME'] = 'DIR:' + self.cache_dir
with get_krb5_context() as k_ctx, get_krb5_cc_default(k_ctx) as cache:
princ = None
try:
# build a principal for 'nobody'
realm = self.realm.encode()
c_realm = ffi.new('char[]', realm)
component = ffi.new('char[]', b'nobody')
p_princ = ffi.new('krb5_principal *')
rc = lib.krb5_build_principal(
k_ctx, p_princ, len(realm), c_realm, component, ffi.NULL)
check_rc(k_ctx, rc)
princ = p_princ[0]
# initialize the default cache with 'nobody' as the default principal
rc = lib.krb5_cc_initialize(k_ctx, cache, princ)
check_rc(k_ctx, rc)
finally:
if princ is not None:
lib.krb5_free_principal(k_ctx, princ)
def addprinc(self, principal: str, password: str):
subprocess.run([

16
ceod/model/LDAPService.py
View File

@ -3,6 +3,7 @@ import grp
import pwd
from typing import Union, List
from flask import g
import ldap3
from zope import component
from zope.interface import implementer
@ -28,18 +29,13 @@ class LDAPService:
self.member_max_id = cfg.get('members_max_id')
self.club_min_id = cfg.get('clubs_min_id')
self.club_max_id = cfg.get('clubs_max_id')
self.sasl_user = None
def set_sasl_user(self, sasl_user: Union[str, None]):
# TODO: store SASL user in flask.g instead
self.sasl_user = sasl_user
def _get_ldap_conn(self, gssapi_bind: bool = True) -> ldap3.Connection:
def _get_ldap_conn(self) -> ldap3.Connection:
kwargs = {'auto_bind': True, 'raise_exceptions': True}
if gssapi_bind:
if 'sasl_user' in g:
kwargs['authentication'] = ldap3.SASL
kwargs['sasl_mechanism'] = ldap3.KERBEROS
kwargs['user'] = self.sasl_user
kwargs['user'] = g.sasl_user
# TODO: cache the connection for a single request
conn = ldap3.Connection(self.ldap_server_url, **kwargs)
return conn
@ -77,12 +73,12 @@ class LDAPService:
return group.ldap3_entry.entry_writable()
def get_user(self, username: str) -> IUser:
conn = self._get_ldap_conn(False)
conn = self._get_ldap_conn()
entry = self._get_readable_entry_for_user(conn, username)
return User.deserialize_from_ldap(entry)
def get_group(self, cn: str) -> IGroup:
conn = self._get_ldap_conn(False)
conn = self._get_ldap_conn()
entry = self._get_readable_entry_for_group(conn, cn)
return Group.deserialize_from_ldap(entry)

2
ceod/model/MailmanService.py
View File

@ -13,7 +13,7 @@ class MailmanService:
def __init__(self):
cfg = component.getUtility(IConfig)
self.base_domain = cfg.get('base_domain')
self.api_base_url = cfg.get('mailman3_api_base_url')
self.api_base_url = cfg.get('mailman3_api_base_url').rstrip('/')
api_username = cfg.get('mailman3_api_username')
api_password = cfg.get('mailman3_api_password')
self.basic_auth = HTTPBasicAuth(api_username, api_password)

4
ceod/transactions/members/AddMemberTransaction.py
View File

@ -4,10 +4,10 @@ from typing import Union, List
from zope import component
from ..AbstractTransaction import AbstractTransaction
from .utils import gen_password
from ceo_common.interfaces import IConfig, IMailService
from ceo_common.logger_factory import logger_factory
from ceod.model import User, Group
import ceod.utils as utils
logger = logger_factory(__name__)
@ -67,7 +67,7 @@ class AddMemberTransaction(AbstractTransaction):
group.add_to_ldap()
yield 'add_group_to_ldap'
password = gen_password()
password = utils.gen_password()
member.add_to_kerberos(password)
yield 'add_user_to_kerberos'

47
ceod/transactions/members/DeleteMemberTransaction.py Normal file
View File

@ -0,0 +1,47 @@
from zope import component
from ..AbstractTransaction import AbstractTransaction
from ceo_common.errors import UserNotSubscribedError
from ceo_common.interfaces import ILDAPService, IConfig
class DeleteMemberTransaction(AbstractTransaction):
"""
Transaction to permanently delete a member and their resources.
This should only be used during testing.
"""
operations = [
'remove_user_from_ldap',
'remove_group_from_ldap',
'remove_user_from_kerberos',
'delete_home_dir',
'unsubscribe_from_mailing_list',
]
def __init__(self, uid: str):
super().__init__()
self.uid = uid
def child_execute_iter(self):
ldap_srv = component.getUtility(ILDAPService)
cfg = component.getUtility(IConfig)
user = ldap_srv.get_user(self.uid)
group = ldap_srv.get_group(self.uid)
new_member_list = cfg.get('mailman3_new_member_list')
user.remove_from_ldap()
yield 'remove_user_from_ldap'
group.remove_from_ldap()
yield 'remove_group_from_ldap'
user.remove_from_kerberos()
yield 'remove_user_from_kerberos'
user.delete_home_dir()
yield 'delete_home_dir'
try:
user.unsubscribe_from_mailing_list(new_member_list)
yield 'unsubscribe_from_mailing_list'
except UserNotSubscribedError:
pass
self.finish('OK')

4
ceod/transactions/members/ModifyMemberTransaction.py
View File

@ -38,12 +38,12 @@ class ModifyMemberTransaction(AbstractTransaction):
user = self.ldap_srv.get_user(self.username)
self.user = user
if self.login_shell:
if self.login_shell is not None:
self.old_login_shell = user.login_shell
user.replace_login_shell(self.login_shell)
yield 'replace_login_shell'
if self.forwarding_addresses:
if self.forwarding_addresses is not None:
self.old_forwarding_addresses = user.get_forwarding_addresses()
user.set_forwarding_addresses(self.forwarding_addresses)
yield 'replace_forwarding_addresses'

27
ceod/transactions/members/ResetPasswordTransaction.py
View File

@ -1,27 +0,0 @@
from zope import component
from ..AbstractTransaction import AbstractTransaction
from .utils import gen_password
from ceo_common.interfaces import ILDAPService
class ResetPasswordTransaction(AbstractTransaction):
"""Transaction to reset a user's password."""
operations = [
'change_password',
]
def __init__(self, username: str):
super().__init__()
self.username = username
self.ldap_srv = component.getUtility(ILDAPService)
def child_execute_iter(self):
user = self.ldap_srv.get_user(self.username)
password = gen_password()
user.change_password(password)
yield 'change_password'
self.finish({'password': password})

2
ceod/transactions/members/__init__.py
View File

@ -1,4 +1,4 @@
from .AddMemberTransaction import AddMemberTransaction
from .ModifyMemberTransaction import ModifyMemberTransaction
from .RenewMemberTransaction import RenewMemberTransaction
from .ResetPasswordTransaction import ResetPasswordTransaction
from .DeleteMemberTransaction import DeleteMemberTransaction

20
ceod/transactions/uwldap/UpdateProgramsTransaction.py
View File

@ -1,20 +0,0 @@
from typing import Union, List
from zope import component
from ..AbstractTransaction import AbstractTransaction
from ceo_common.interfaces import ILDAPService
class UpdateProgramsTransaction(AbstractTransaction):
"""Transaction to sync the 'program' attribute in CSC LDAP with UW LDAP."""
def __init__(self, members: Union[List[str], None]):
super().__init__()
self.members = members
self.ldap_srv = component.getUtility(ILDAPService)
def child_execute_iter(self):
users_updated = self.ldap_srv.update_programs(members=self.members)
yield 'update_programs'
self.finish(users_updated)

1
ceod/transactions/uwldap/__init__.py
View File

@ -1 +0,0 @@
from .UpdateProgramsTransaction import UpdateProgramsTransaction

0
ceod/transactions/members/utils.py → ceod/utils.py
View File

14
gen_cred.py Executable file
View File

@ -0,0 +1,14 @@
#!/usr/bin/env python3
from base64 import b64encode
import sys
from ceo_common.krb5.utils import get_fwd_tgt
if len(sys.argv) != 2:
print(f'Usage: {sys.argv[0]} <ceod hostname>', file=sys.stderr)
sys.exit(1)
b = get_fwd_tgt('ceod/' + sys.argv[1])
with open('cred', 'wb') as f:
f.write(b64encode(b))

26
tests/MockMailmanServer.py
View File

@ -13,7 +13,13 @@ class MockMailmanServer:
self.runner = web.AppRunner(self.app)
self.loop = asyncio.new_event_loop()
self.subscriptions = []
# add more as necessary
self.subscriptions = {
'csc-general': [],
'exec': [],
'syscom': [],
'syscom-alerts': [],
}
def _start_loop(self):
asyncio.set_event_loop(self.loop)
@ -32,18 +38,28 @@ class MockMailmanServer:
async def subscribe(self, request):
body = await request.post()
subscriber = body['subscriber']
if subscriber in self.subscriptions:
list_id = body['list_id']
list_id = list_id[:list_id.index('.')]
if list_id not in self.subscriptions:
return web.json_response({
'description': 'No such list',
}, status=400)
subscribers = self.subscriptions[list_id]
if subscriber in subscribers:
return web.json_response({
'description': 'user is already subscribed',
}, status=409)
self.subscriptions.append(subscriber)
subscribers.append(subscriber)
return web.json_response({'status': 'OK'})
async def unsubscribe(self, request):
subscriber = request.match_info['address']
if subscriber not in self.subscriptions:
list_id = request.match_info['mailing_list']
list_id = list_id[:list_id.index('@')]
subscribers = self.subscriptions.get(list_id, [])
if subscriber not in subscribers:
return web.json_response({
'description': 'user is not subscribed',
}, status=404)
self.subscriptions.remove(subscriber)
subscribers.remove(subscriber)
return web.json_response({'status': 'OK'})

200
tests/ceod/api/test_members.py
View File

@ -1,23 +1,205 @@
import pwd
import grp
from unittest.mock import patch
import ldap3
import pytest
import ceod.utils as utils
def test_members_get_user(client):
def test_api_user_not_found(client):
status, data = client.get('/api/members/no_such_user')
assert status == 404
assert data['error'] == 'user not found'
@pytest.fixture(scope='session')
def create_user_resp(client):
return client.post('/api/members', json={
'uid': 'test_jdoe',
'cn': 'John Doe',
def mocks_for_create_user():
with patch.object(utils, 'gen_password') as gen_password_mock, \
patch.object(pwd, 'getpwuid') as getpwuid_mock, \
patch.object(grp, 'getgrgid') as getgrgid_mock:
gen_password_mock.return_value = 'krb5'
# Normally, if getpwuid or getgrgid do *not* raise a KeyError,
# then LDAPService will skip that UID. Therefore, by raising a
# KeyError, we are making sure that the UID will *not* be skipped.
getpwuid_mock.side_effect = KeyError()
getgrgid_mock.side_effect = KeyError()
yield
@pytest.fixture(scope='session')
def create_user_resp(client, mocks_for_create_user):
status, data = client.post('/api/members', json={
'uid': 'test_1',
'cn': 'Test One',
'program': 'Math',
'terms': ['s2021'],
})
assert status == 200
assert data[-1]['status'] == 'completed'
yield status, data
status, data = client.delete('/api/members/test_1')
assert status == 200
assert data[-1]['status'] == 'completed'
# def test_create_user(create_user_resp):
# status, data = create_user_resp
# assert status == 200
# # TODO: check response contents
@pytest.fixture(scope='session')
def create_user_result(create_user_resp):
# convenience method
_, data = create_user_resp
return data[-1]['result']
def test_api_create_user(cfg, create_user_resp):
_, data = create_user_resp
min_uid = cfg.get('members_min_id')
expected = [
{"status": "in progress", "operation": "add_user_to_ldap"},
{"status": "in progress", "operation": "add_group_to_ldap"},
{"status": "in progress", "operation": "add_user_to_kerberos"},
{"status": "in progress", "operation": "create_home_dir"},
{"status": "in progress", "operation": "set_forwarding_addresses"},
{"status": "in progress", "operation": "send_welcome_message"},
{"status": "in progress", "operation": "subscribe_to_mailing_list"},
{"status": "completed", "result": {
"cn": "Test One",
"uid": "test_1",
"uid_number": min_uid,
"gid_number": min_uid,
"login_shell": "/bin/bash",
"home_directory": "/tmp/test_users/test_1",
"is_club": False,
"program": "Math",
"terms": ["s2021"],
"forwarding_addresses": [],
"password": "krb5"
}},
]
assert data == expected
def test_api_next_uid(cfg, client, create_user_result):
min_uid = cfg.get('members_min_id')
_, data = client.post('/api/members', json={
'uid': 'test_2',
'cn': 'Test Two',
'program': 'Math',
'terms': ['s2021'],
})
assert data[-1]['status'] == 'completed'
result = data[-1]['result']
try:
assert result['uid_number'] == min_uid + 1
assert result['gid_number'] == min_uid + 1
finally:
client.delete('/api/members/test_2')
def test_api_get_user(cfg, client, create_user_result):
old_data = create_user_result
uid = old_data['uid']
del old_data['password']
status, data = client.get(f'/api/members/{uid}')
assert status == 200
assert data == old_data
def test_api_patch_user(client, create_user_result):
data = create_user_result
uid = data['uid']
prev_login_shell = data['login_shell']
prev_forwarding_addresses = data['forwarding_addresses']
# replace login shell
new_shell = '/bin/sh'
status, data = client.patch(
f'/api/members/{uid}', json={'login_shell': new_shell})
assert status == 200
expected = [
{"status": "in progress", "operation": "replace_login_shell"},
{"status": "completed", "result": "OK"},
]
assert data == expected
# replace forwarding addresses
new_forwarding_addresses = [
'test@example1.internal', 'test@example2.internal',
]
status, data = client.patch(
f'/api/members/{uid}', json={
'forwarding_addresses': new_forwarding_addresses
}
)
assert status == 200
expected = [
{"status": "in progress", "operation": "replace_forwarding_addresses"},
{"status": "completed", "result": "OK"},
]
assert data == expected
# retrieve the user and make sure that both fields were changed
status, data = client.get(f'/api/members/{uid}')
assert status == 200
assert data['login_shell'] == new_shell
assert data['forwarding_addresses'] == new_forwarding_addresses
# replace (restore) both
status, data = client.patch(
f'/api/members/{uid}', json={
'login_shell': prev_login_shell,
'forwarding_addresses': prev_forwarding_addresses
}
)
assert status == 200
expected = [
{"status": "in progress", "operation": "replace_login_shell"},
{"status": "in progress", "operation": "replace_forwarding_addresses"},
{"status": "completed", "result": "OK"},
]
assert data == expected
def test_api_renew_user(cfg, client, create_user_result, ldap_conn):
data = create_user_result
uid = data['uid']
old_terms = data['terms']
old_non_member_terms = data.get('non_member_terms', [])
new_terms = ['f2021']
status, data = client.post(
f'/api/members/{uid}/renew', json={'terms': new_terms})
assert status == 200
new_non_member_terms = ['w2022', 's2022']
status, data = client.post(
f'/api/members/{uid}/renew', json={'non_member_terms': new_non_member_terms})
assert status == 200
# check that the changes were applied
_, data = client.get(f'/api/members/{uid}')
assert data['terms'] == old_terms + new_terms
assert data['non_member_terms'] == old_non_member_terms + new_non_member_terms
# cleanup
base_dn = cfg.get('ldap_users_base')
dn = f'uid={uid},{base_dn}'
changes = {
'term': [(ldap3.MODIFY_REPLACE, old_terms)],
'nonMemberTerm': [(ldap3.MODIFY_REPLACE, old_non_member_terms)],
}
ldap_conn.modify(dn, changes)
def test_api_reset_password(client, create_user_result):
uid = create_user_result['uid']
with patch.object(utils, 'gen_password') as gen_password_mock:
gen_password_mock.return_value = 'new_password'
status, data = client.post(f'/api/members/{uid}/pwreset')
assert status == 200
assert data['password'] == 'new_password'
# cleanup
status, data = client.post(f'/api/members/{uid}/pwreset')
assert status == 200
assert data['password'] == 'krb5'

4
tests/ceod/model/test_group.py
View File

@ -3,7 +3,7 @@ import pytest
from ceo_common.errors import GroupNotFoundError
def test_group_add_to_ldap(simple_group, ldap_srv):
def test_group_add_to_ldap(simple_group, ldap_srv, g_admin):
group = simple_group
group.add_to_ldap()
@ -15,7 +15,7 @@ def test_group_add_to_ldap(simple_group, ldap_srv):
ldap_srv.get_group(group.cn)
def test_group_members(ldap_group, ldap_srv):
def test_group_members(ldap_group, ldap_srv, g_admin):
group = ldap_group
group.add_member('member1')

1
tests/ceod/model/test_mail.py
View File

@ -1,5 +1,6 @@
def test_welcome_message(cfg, mock_mail_server, mail_srv, simple_user):
base_domain = cfg.get('base_domain')
mock_mail_server.messages.clear()
mail_srv.send_welcome_message_to(simple_user)
msg = mock_mail_server.messages[0]
assert msg['from'] == f'exec@{base_domain}'

10
tests/ceod/model/test_user.py
View File

@ -7,7 +7,7 @@ from ceo_common.errors import UserNotFoundError, UserAlreadyExistsError
from ceod.model import User
def test_user_add_to_ldap(cfg, ldap_srv, simple_user):
def test_user_add_to_ldap(cfg, ldap_srv, simple_user, g_admin):
user = simple_user
min_id = cfg.get('members_min_id')
user.add_to_ldap()
@ -23,7 +23,7 @@ def test_user_add_to_ldap(cfg, ldap_srv, simple_user):
ldap_srv.get_user(user.uid)
def test_club_add_to_ldap(cfg, ldap_srv, simple_club):
def test_club_add_to_ldap(cfg, ldap_srv, simple_club, g_admin):
club = simple_club
min_id = cfg.get('clubs_min_id')
club.add_to_ldap()
@ -74,7 +74,7 @@ def test_user_forwarding_addresses(cfg, ldap_user):
assert not os.path.isdir(user.home_directory)
def test_user_terms(ldap_user, ldap_srv):
def test_user_terms(ldap_user, ldap_srv, g_admin):
user = ldap_user
user.add_terms(['f2021'])
@ -86,7 +86,7 @@ def test_user_terms(ldap_user, ldap_srv):
assert ldap_srv.get_user(user.uid).non_member_terms == user.non_member_terms
def test_user_positions(ldap_user, ldap_srv):
def test_user_positions(ldap_user, ldap_srv, g_admin):
user = ldap_user
user.add_position('treasurer')
@ -108,7 +108,7 @@ def test_user_change_password(krb_user):
user.change_password('new_password')
def test_login_shell(ldap_user, ldap_srv):
def test_login_shell(ldap_user, ldap_srv, g_admin):
user = ldap_user
user.replace_login_shell('/bin/sh')

7
tests/ceod/model/test_uwldap.py
View File

@ -1,7 +1,5 @@
import ldap3
from tests.conftest import get_ldap_conn
def test_uwldap_get(uwldap_srv, uwldap_user):
retrieved_user = uwldap_srv.get_user(uwldap_user.uid)
@ -11,11 +9,12 @@ def test_uwldap_get(uwldap_srv, uwldap_user):
assert uwldap_srv.get_user('no_such_user') is None
def test_ldap_updateprograms(cfg, ldap_srv, uwldap_srv, ldap_user, uwldap_user):
def test_ldap_updateprograms(
cfg, ldap_conn, ldap_srv, uwldap_srv, ldap_user, uwldap_user, g_admin):
# sanity check
assert ldap_user.uid == uwldap_user.uid
# modify the user's program in UWLDAP
conn = get_ldap_conn(cfg.get('uwldap_server_url'))
conn = ldap_conn
base_dn = cfg.get('uwldap_base')
dn = f'uid={uwldap_user.uid},{base_dn}'
changes = {'ou': [(ldap3.MODIFY_REPLACE, ['New Program'])]}

9
tests/ceod_dev.ini
View File

@ -7,6 +7,7 @@ admin_host = phosphoric-acid
# this is the host with NFS no_root_squash
fs_root_host = phosphoric-acid
mailman_host = mail
krb5_cache_dir = /run/ceod/krb5_cache
use_https = false
port = 9987
@ -43,3 +44,11 @@ api_base_url = http://localhost:8001/3.1
api_username = restadmin
api_password = mailman3
new_member_list = csc-general
[auxiliary groups]
syscom = office,staff,adm,src
office = cdrom,audio,video,www
[auxiliary mailing lists]
syscom = syscom,syscom-alerts,syscom-moderators,mirror
exec = exec,exec-moderators

1
tests/ceod_test_local.ini
View File

@ -5,6 +5,7 @@ base_domain = csclub.internal
admin_host = phosphoric-acid
fs_root_host = phosphoric-acid
mailman_host = mail
krb5_cache_dir = /tmp/ceod_test_krb5_cache
use_https = false
port = 9987

70
tests/conftest.py
View File

@ -1,7 +1,10 @@
import importlib.resources
import os
import shutil
import subprocess
import tempfile
import flask
import ldap3
import pytest
import socket
@ -30,16 +33,16 @@ def cfg():
@pytest.fixture(scope='session')
def krb_srv(cfg):
# we need to be root to read the keytab
# TODO: create temporary Kerberos database using kdb5_util.
# We need to be root to read the keytab
assert os.geteuid() == 0
# this dance again... ugh
if socket.gethostname() == cfg.get('ceod_admin_host'):
principal = 'ceod/admin'
else:
principal = 'ceod/' + socket.getfqdn()
cache_dir = '/tmp/ceod_test/krb5_cache'
shutil.rmtree(cache_dir, ignore_errors=True)
krb = KerberosService(principal, cache_dir)
cache_dir = cfg.get('ceod_krb5_cache_dir')
krb = KerberosService(principal)
component.provideUtility(krb, IKerberosService)
yield krb
shutil.rmtree(cache_dir)
@ -55,15 +58,35 @@ def delete_subtree(conn: ldap3.Connection, base_dn: str):
pass
def get_ldap_conn(server_url: str):
return ldap3.Connection(
server_url, auto_bind=True, raise_exceptions=True,
authentication=ldap3.SASL, sasl_mechanism=ldap3.KERBEROS)
@pytest.fixture(scope='session')
def ceod_admin_creds(cfg, krb_srv):
"""
Acquire credentials for ceod/admin and store them
in the default ccache.
"""
subprocess.run(
['kinit', '-k', cfg.get('ldap_admin_principal')],
check=True,
)
@pytest.fixture(scope='session')
def ldap_srv(cfg, krb_srv):
conn = get_ldap_conn(cfg.get('ldap_server_url'))
def ldap_conn(cfg, ceod_admin_creds) -> ldap3.Connection:
# Assume that the same server URL is being used for the CSC
# and UWLDAP during the tests.
cfg = component.getUtility(IConfig)
server_url = cfg.get('ldap_server_url')
# sanity check
assert server_url == cfg.get('uwldap_server_url')
return ldap3.Connection(
server_url, auto_bind=True, raise_exceptions=True,
authentication=ldap3.SASL, sasl_mechanism=ldap3.KERBEROS,
user=cfg.get('ldap_admin_principal'))
@pytest.fixture(scope='session')
def ldap_srv(cfg, krb_srv, ldap_conn):
conn = ldap_conn
users_base = cfg.get('ldap_users_base')
groups_base = cfg.get('ldap_groups_base')
@ -119,8 +142,8 @@ def mailman_srv(mock_mailman_server, cfg, http_client):
@pytest.fixture(scope='session')
def uwldap_srv(cfg, ldap_srv):
conn = get_ldap_conn(cfg.get('uwldap_server_url'))
def uwldap_srv(cfg, ldap_conn):
conn = ldap_conn
base_dn = cfg.get('uwldap_base')
ou = base_dn.split(',', 1)[0].split('=')[1]
@ -167,6 +190,21 @@ def app(
return app
@pytest.fixture
def g_admin(cfg, ceod_admin_creds, app):
"""
Store the creds for ceod/admin in flask.g.
This fixture should be used any time LDAP is modified via the LDAPService.
"""
admin_principal = cfg.get('ldap_admin_principal')
with app.app_context():
try:
flask.g.sasl_user = admin_principal
yield
finally:
flask.g.pop('sasl_user')
@pytest.fixture
def simple_user():
return User(
@ -187,7 +225,7 @@ def simple_club():
@pytest.fixture
def ldap_user(simple_user):
def ldap_user(simple_user, g_admin):
simple_user.add_to_ldap()
yield simple_user
simple_user.remove_from_ldap()
@ -209,15 +247,15 @@ def simple_group():
@pytest.fixture
def ldap_group(simple_group):
def ldap_group(simple_group, g_admin):
simple_group.add_to_ldap()
yield simple_group
simple_group.remove_from_ldap()
@pytest.fixture
def uwldap_user(cfg, uwldap_srv):
conn = get_ldap_conn(cfg.get('uwldap_server_url'))
def uwldap_user(cfg, uwldap_srv, ldap_conn):
conn = ldap_conn
base_dn = cfg.get('uwldap_base')
user = UWLDAPRecord(
uid='test_jdoe',

60
tests/conftest_ceod_api.py
View File

@ -1,59 +1,82 @@
from base64 import b64encode
import json
import socket
import subprocess
import tempfile
from flask.testing import FlaskClient
import gssapi
import pytest
from requests import Request
from requests_gssapi import HTTPSPNEGOAuth
from zope import component
from ceo_common.interfaces import IConfig
from ceo_common.krb5.utils import get_fwd_tgt
__all__ = ['client']
@pytest.fixture(scope='session')
def client(app):
app_client = app.test_client()
return CeodTestClient(app_client)
with tempfile.TemporaryDirectory() as cache_dir:
yield CeodTestClient(app_client, cache_dir)
class CeodTestClient:
def __init__(self, app_client: FlaskClient):
cfg = component.getUtility(IConfig)
def __init__(self, app_client: FlaskClient, cache_dir: str):
self.client = app_client
self.admin_principal = cfg.get('ldap_admin_principal')
self.syscom_principal = 'ctdalek'
# this is only used for the HTTPSNEGOAuth
self.base_url = f'http://{socket.getfqdn()}'
self.cached_auth = {}
# keep a list of all of the principals for which we acquired a TGT
self.principals = []
# this is where we'll store the credentials for each principal
self.ccache = 'DIR:' + cache_dir
def get_auth(self, principal):
if principal in self.cached_auth:
return self.cached_auth[principal]
name = gssapi.Name(principal)
creds = gssapi.Credentials(name=name, usage='initiate')
creds = gssapi.Credentials(
name=name,
usage='initiate',
store={'ccache': self.ccache},
)
auth = HTTPSPNEGOAuth(
opportunistic_auth=True,
target_name='ceod',
creds=creds,
)
self.cached_auth[principal] = auth
return auth
def get_headers(self, principal):
# method doesn't matter here because we just need the headers
if principal not in self.principals:
# Acquire the initial TGT
subprocess.run(
['kinit', '-c', self.ccache, principal],
text=True, input='krb5',
check=True, stdout=subprocess.DEVNULL)
self.principals.append(principal)
# Get the Authorization header (SPNEGO).
# The method doesn't matter here because we just need to extract
# the header using req.prepare().
req = Request('GET', self.base_url, auth=self.get_auth(principal))
return req.prepare().headers.items()
headers = list(req.prepare().headers.items())
# Get the X-KRB5-CRED header (forwarded TGT).
cred = b64encode(
get_fwd_tgt('ceod/' + socket.getfqdn(), self.ccache)
).decode()
headers.append(('X-KRB5-CRED', cred))
return headers
def request(self, method, path, principal, **kwargs):
if principal is None:
principal = self.admin_principal
principal = self.syscom_principal
resp = self.client.open(
path, method=method, headers=self.get_headers(principal), **kwargs)
status = int(resp.status.split(' ', 1)[0])
try:
if resp.headers['content-type'] == 'application/json':
data = json.loads(resp.data)
except json.JSONDecodeError:
data = resp.data.decode()
else:
data = [json.loads(line) for line in resp.data.splitlines()]
return status, data
def get(self, path, principal=None, **kwargs):
@ -62,5 +85,8 @@ class CeodTestClient:
def post(self, path, principal=None, **kwargs):
return self.request('POST', path, principal, **kwargs)
def patch(self, path, principal=None, **kwargs):
return self.request('PATCH', path, principal, **kwargs)
def delete(self, path, principal=None, **kwargs):
return self.request('DELETE', path, principal, **kwargs)