source: tools/suggest/suggest.py @ c3e4516

#!/usr/local/bin/python

import ABAC
import Creddy

import os
import re
import copy
import base64

from tempfile import mkdtemp
from shutil import rmtree

class proof:
    def __init__(self, name, prover, role, principal, creds):
        self.name = name
        self.prover = prover
        self.role = role
        self.principal = principal
        self.ctxt = ABAC.Context()
        self.keyid_to_cn = { }
        self.cn_to_keyid = { }
        attrs = []
        try:
            d = mkdtemp()
            for c in creds:
                cc = self.pem_to_der(c)
                succ = self.ctxt.load_id_chunk(cc) 
                if succ == ABAC.ABAC_CERT_SUCCESS: 
                    try:
                        fn = os.path.join(d, 'file.pem')
                        f = open(fn, 'w')
                        f.write(cc)
                        f.close()
                        cid = Creddy.ID(fn)
                        base_cn = cn = re.sub('_ID.pem$','',cid.cert_filename())
                        i = 0
                        while cn in self.cn_to_keyid:
                            cn = '%s%03d' % (base_cn, i)
                            i += 1
                        self.cn_to_keyid[cn] = cid.keyid()
                        self.keyid_to_cn[cid.keyid()] = cn
                    except EnvironmentError, e:
                        print >>sys.stderr, '%s: %s' % (e.filename, e.strerror)
                else:
                    attrs.append(cc)
            for c in attrs:
                self.ctxt.load_attribute_chunk(c) 
        finally:
            rmtree(d)


    @staticmethod
    def pem_to_der(c):
        pat = '-----BEGIN CERTIFICATE-----(.*)-----END CERTIFICATE'
        m = re.match(pat, c, re.DOTALL)
        if m: return base64.b64decode(m.group(1))
        else: return c

    def replace_keyids(self, s):
        for k, v in self.keyid_to_cn.items():
            s = re.sub(k, v, s)
        return s

    def __str__(self):
        s = 'Name: %s\n' % self.name
        s += 'Prover: %s\n' % self.prover
        s += 'Principal: %s\n' % self.principal
        s += 'Role: %s\n' % self.role
        s += 'Creds: \n'
        for c in self.ctxt.credentials():
            s += self.replace_keyids(
                    '%s <- %s\n' % ( c.head().string(), c.tail().string()))
        return s

def read_proofs(fn):
    proofs = []
    try:
        f = open(fn, 'r')
        creds = []
        for line in f:
            line = line.strip()
            if line == '<proof>':
                prover = None
                principal = None
                role = None
                creds = []
            elif line == '</proof>' :
                p = proof(name, prover, role, principal, creds)
                ok, pp = p.ctxt.query(role, principal)
                if not ok: proofs.append(p)
           
            m = re.match('<comment>(.*)</comment>', line)
            if m is not None:
                name = m.group(1)

            m = re.match('<principal>([0-9a-f]+)</principal>', line)
            if m is not None:
                principal = m.group(1)

            m = re.match('<prover>([0-9a-f]+)</prover>', line)
            if m is not None:
                prover = m.group(1)

            m = re.match('<attribute>(.*)</attribute>', line)
            if m is not None:
                role = m.group(1)

            m = re.match('<credential>(.*)</credential>', line)
            if m is not None:
                creds.append(base64.b64decode(m.group(1)))
        f.close()
    except EnvironmentError, e:
        return None
    return proofs

def interpret_proof(p):
    roles = set()
    direct_roles = {}
    groles = set()
    principals = set()
    goals = set()
    attrs = set()
    ok, proof = p.ctxt.query(p.role, p.principal)
    for c in p.ctxt.credentials():
        role = c.tail()
        if role.is_principal():
            if role.string() != p.principal: 
                principals.add(role.string())
            else:
                assigner, r = c.head().string().split('.')
                direct_roles[r] = assigner
        else: 
            r = role.string()
            for s in r.split('&'):
                roles.add(s.strip())
            groles.add(r)

        role = c.head()
        roles.add(role.string())
        groles.add(role.string())

    for r in groles:
        ok, proof =  p.ctxt.query(p.role, r)
        if ok :
            goals.add(r)
    for r in roles:
        ok, proof = p.ctxt.query(r, p.principal)
        if ok:
            attrs.add(r)

    split_goals = [ [s.strip() for s in g.split('&')] for g in goals ]
    plans = []
    for sg in split_goals:
        p = []
        for g in sg:
            if g.count('.') == 2:
                # linking role
                pr, rr, lr = g.split('.')
                if lr in direct_roles:
                    p.append('add %s.%s to %s' % (pr, rr, direct_roles[lr]))
                else:
                    p.append('someone with %s.%s must delegate %s to %s' % \
                            (pr, rr, lr, p.principal))
            elif g.count('.') == 1:
                p.append('add %s to %s' % (g, principal))
        plans.append('\n'.join(p))

    return plans 


proofs = read_proofs('./auth.log')
plans = interpret_proof(proofs[0])
for p in proofs:
    plans = interpret_proof(p)
    print "%s needs to have %s" % (p.principal, p.role)
    print "accompilsh this by: "
    first = True
    for pl in plans:
        if not first: print '---'
        else: first = False
        print pl