Produits Entr'ouvert » Hobo

from optparse import make_option

from django.core.management.base import BaseCommand

from .. import role_extractors

class Command(BaseCommand):

    option_list = BaseCommand.option_list + (

        make_option('--slug', action='store', dest='slug'),

        make_option('--key', action='store', dest='key'),

        make_option('--orig', action='store', dest='orig'),

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

    )

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

    requires_system_checks = False

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

        role_extractor = role_extractors.WcsRoleExtractor(

            slug=options['slug'],

            key=options['key'],

            orig=options['orig'],

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

        )

        role_extractor.extract()

import logging

import requests

import urllib

from django.utils.text import slugify

from authentic2.saml.models import LibertyProvider

from authentic2.model import ServiceRole, ServiceRoleAttribute

from . import signature

class RoleExtractor(object):

    def __init__(self, slug, delete=False):

        self.slug = slug

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

        self.delete = delete

    def extract_impl(self):

        pass

    def extract(self):

        seen_ids = set()

        for role_tpl in self.extract_impl():

            defaults = {

                'name': role_tpl['name'],

                'slug': role_tpl.get('slug', slugify(role_tpl['name'])),

            }

            seen_ids.add(role_tpl['seen_ids'])

            role, created = ServiceRole.objects.get_or_create(

                    service=self.service,

                    external_id=role_tpl['external_id'],

                    defaults=defaults)

            ServiceRoleAttribute.objects.get_or_create(

                    service_role=role,

                    name=role_tpl['attribute_name'],

                    kind='string',

                    value=role_tpl['external_id'])

            if created:

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

                        '%s', role_tpl['external_id'], role_tpl['name'],

                        self.slug)

            self.logger

            if not created:

                # Update name if it has changed

                if role.name != role_tpl['name']:

                    role.name = role_tpl['name']

                    role.save()

        if self.delete:

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

                .exclude(external_id__in=list(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()

class WcsRoleExtractor(RoleExtractor):

    def __init__(self, *args, **kwargs):

        self.key = kwargs.pop('key')

        self.orig = kwargs.pop('orig')

        self.attribute_name = kwargs.pop('attribute_name', 'role_id')

        super(WcsRoleExtractor, self).__init__(*args, **kwargs)

        self.service = LibertyProvider.objects.get(slug=self.slug)

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

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

    def extract_impl(self):

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

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

        response = requests.get(signed_url)

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

            yield {

                'attribute_name': self.attribute_name,

                'external_id': str(role['id']),

                'name': role['text'],

            }

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()