Produits Entr'ouvert » Hobo

 - authentic2

authentic2 instances will be deployed using "/usr/bin/authentic2-ctl" by

default, this command can be adapted in the AUTHENTIC_MANAGE_COMMAND setting.

It should be run with the same rights as the authentic2 process (redefine the

command to use sudo if necessary).

The agent also provide a commands to import roles from w.c.s named

import-wcs-roles. It computes the web-service credentials from the hobo.json

and use the email of the oldest superuser. Cron job can be created for calling

this command when regular synchronization of roles with your w.c.s.  instances

is needed. The sole option named "--delete" indicate if you want to delete

stale roles, default is to not delete them.

import logging

import requests

import urllib

import urlparse

import hashlib

from optparse import make_option

from django.utils.text import slugify

from django.core.management.base import BaseCommand

from django.contrib.auth import get_user_model

from authentic2.saml.models import LibertyProvider

from authentic2.a2_rbac.models import Role, RoleAttribute

from hobo import signature

from hobo.multitenant.middleware import TenantMiddleware

from tenant_schemas.utils import tenant_context

from .. import role_extractors

class WcsRoleImporter(object):

    def __init__(self, liberty_provider, key, orig, email,

            attribute_name='role_id', delete=False):

        self.service = liberty_provider

        self.slug = liberty_provider.slug

        self.key = key

        self.orig = orig

        self.email = email

        self.attribute_name = attribute_name

        self.delete = delete

        assert 'saml/metadata' in self.service.entity_id

        self.wcs_url = self.service.entity_id.split('saml/metadata')[0]

        self.logger = logging.getLogger('%s.%s' % (__name__, self.__class__.__name__))

        self.seen_ids = set()

    def import_roles(self):

        for role_tpl in self.get_roles():

            self.seen_ids.add(role_tpl.external_id)

            self.create_role(role_tpl)

        if self.delete:

            self.delete_dead_roles()

    def create_role(self, role_tpl):

        defaults = {

            'name': role_tpl.name,

            # w.c.s. will always provide a slug but for other services we do

            # not know

            'slug': role_tpl.slug or slugify(role_tpl.name),

        }

        # search role by external id, create if not found

        role, created = Role.objects.get_or_create(

                service=self.service,

                external_id=role_tpl.external_id,

                defaults=defaults)

        role_attribute, ra_created = RoleAttribute.objects.get_or_create(

                role=role,

                name=self.attribute_name,

                kind='string',

                defaults={

                    'value': role_tpl.external_id

                })

        if created:

            self.logger.info('imported new role %r(%r) from service '

                    '%s', role.external_id, role.name, self.slug)

        # update role attribute value if it has changed

        if not ra_created:

            if role_attribute.value != role_tpl.external_id:

                role_attribute.value = role_tpl.external_id

                role_attribute.save()

        # update role name if has changed

        if not created:

            # Update name and slug if they have changed

            if role.name != role_tpl.name:

                role.name = role_tpl.name

                role.save()

    def delete_dead_roles(self):

        '''Deletes service roles whose id is not in self.seen_ids'''

        qs = Role.objects.filter(service=self.service) \

            .exclude(external_id__in=list(self.seen_ids))

        for role in qs:

            self.logger.info('deleted dead role %r(%r) from service '

                    '%s', role.external_id, role.slug, self.slug)

        qs.delete()

    def get_roles(self):

        '''Get w.c.s. from its roles web-service by sending a signed GET request'''

        url = self.wcs_url + 'roles?%s' % urllib.urlencode({'orig': self.orig, 'email': self.email})

        signed_url = signature.sign_url(url, self.key)

        response = requests.get(signed_url)

        for role in response.json()['data']:

            yield Role(

                    name=role['text'],

                    external_id=str(role['slug']),

                    slug=str(role['slug']))

class Command(BaseCommand):

    option_list = BaseCommand.option_list + (

        make_option('--delete', action='store_true', dest='delete'),

    )

    help = "Import W.C.S. roles"

    requires_system_checks = False

    def handle(self, *args, **options):

        # traverse list of tenants

        for tenant in TenantMiddleware.get_tenants():

            with tenant_context(tenant):

                self.handle_tenant(tenant, **options)

    def handle_tenant(self, tenant, **options):

        # extract informations on deployed w.c.s. instances from hobo.json

        hobo_json_path = os.path.join(tenant.get_directory(), 'hobo.json')

        if not os.path.exists(hobo_json_path):

            print 'skipping %s, no hobo.json found' % tenant

            return

        hobo_environment = json.load(open(hobo_json_path))

        # compute our credentials from our hobo configuration

        me = [x for x in hobo_environment['services'] if x.get('this')]

        if not me:

            print 'skipping %s, self services is not marked' % tenant

            return

        orig = urlparse.urlsplit(me['base_url']).netloc

        key = hashlib.sha1(orig+me['secret_key']).hexdigest()

        # FIXME: get mail of the oldest superuser, could we do better ?

        User = get_user_model()

        email = User.objects.order_by('id').filter(email__contains='@', is_superuser=True)[0].email

        for service in hobo_environment['services']:

            if not service.get('saml-sp-metadata-url'):

                continue

            liberty_provider = LibertyProvider.objects.get(entity_id=service['saml-sp-metadata-url'])

            importer = WcsRoleImporter(

                liberty_provider=liberty_provider,

                key=key,

                orig=orig,

                email=email,

                delete=options.get('delete', False),

            )

            importer.import_roles()

import base64

import hmac

import hashlib

import datetime

import urllib

import urllib2

import urlparse

import random

def sign_url(url, key, algo='sha256', timestamp=None, nonce=None):

    parsed = urlparse.urlparse(url)

    new_query = sign_query(parsed.query, key, algo, timestamp, nonce)

    return urlparse.urlunparse(parsed[:4] + (new_query,) + parsed[5:])

def sign_query(query, key, algo='sha256', timestamp=None, nonce=None):

    if timestamp is None:

        timestamp = datetime.datetime.utcnow()

    timestamp = timestamp.strftime('%Y-%m-%dT%H:%M:%SZ')

    if nonce is None:

        nonce = hex(random.getrandbits(128))[2:-1]

    new_query = query

    if new_query:

        new_query += '&'

    new_query += urllib.urlencode((

        ('algo', algo),

        ('timestamp', timestamp),

        ('nonce', nonce)))

    signature = base64.b64encode(sign_string(new_query, key, algo=algo))

    new_query += '&signature=' + urllib.quote(signature)

    return new_query

def sign_string(s, key, algo='sha256', timedelta=30):

    digestmod = getattr(hashlib, algo)

    hash = hmac.HMAC(key, digestmod=digestmod, msg=s)

    return hash.digest()