source: fedd/federation/proof.py @ ee7f7e4

axis_examplecompt_changesinfo-ops
Last change on this file since ee7f7e4 was e83f2f2, checked in by Ted Faber <faber@…>, 14 years ago

Move proofs around. Lots of changes, including fault handling.

  • Property mode set to 100644
File size: 3.8 KB
Line 
1#/usr/local/bin/python
2
3
4import ABAC
5import sys
6import os, os.path
7
8from string import join
9from base64 import b64encode, b64decode
10from xml.parsers.expat import ParserCreate
11
12class proof:
13    """
14    Class to make manipulating proofs a little simpler.  Keeps track of the
15    target and does some of the credential formatting.
16    """
17    def __init__(self, me, p, a, c):
18        """
19        P is the principal (string), a is the attribute to prove (string)
20        and c is the list of abac certificates that prove (or partially
21        prove) the connection.
22        """
23        self.prover = "%s" % me
24        self.principal = "%s" % p
25        self.attribute = a
26        self.creds = c or []
27
28    def creds_to_certs(self):
29        ids = set([c.issuer_cert() for c in self.creds])
30        attrs = set([c.attribute_cert() for c in self.creds])
31        return list(ids) + list(attrs)
32
33    def __str__(self):
34        """
35        User friendly output format
36        """
37        rv = "prover: %s\nprincipal: %s\nattribute: %s" % \
38                (self.prover, self.principal, self.attribute)
39        if self.creds:
40            rv += '\n'
41            rv += join(["%s <- %s"%(c.head().string(), c.tail().string()) \
42                for c in self.creds], '\n')
43        return rv
44
45    def to_dict(self):
46        """
47        Dict with the credentials encoded as X.509 certificates
48        """
49
50        return { 
51                'prover': self.prover,
52                'principal': self.principal,
53                'attribute': self.attribute,
54                'credential': self.creds_to_certs(),
55                }
56    def to_context(self):
57        """
58        Return an ABAC context containing just the credentials in the proof
59        """
60        ctxt = ABAC.Context()
61        for c in self.creds_to_certs():
62            if ctxt.load_id_chunk(c) != ABAC.ABAC_CERT_SUCCESS:
63                ctxt.load_attribute_chunk(c)
64        return ctxt
65
66    def to_xml(self):
67        """
68        Convert the proof to a simple XML format
69        """
70        rv = "<proof>\n"
71        rv += "<prover>%s</prover>\n" % self.prover
72        rv += "<principal>%s</principal>\n" % self.principal
73        rv += "<attribute>%s</attribute>\n" % self.attribute
74        for c in self.creds_to_certs(): 
75            rv += "<credential>%s</credential>\n" % b64encode(c)
76        rv += "</proof>"
77        return rv
78
79    def save_to_xml(self, fn, mode='w'):
80        """
81        Write an XML reprsentation to fn.  Mode is 'w' or 'a' to overwrite or
82        append to the file.
83        """
84        f = open(fn, mode)
85        print >>f, self.to_xml()
86        f.close()
87
88    @staticmethod
89    def from_xml(filename=None, file=None, string=None):
90        """
91        Create a proof from the XML representation
92        """
93        class parse_proof:
94            def element_end(self, name):
95                if name =='proof':
96                    self.proof = proof(self.me, self.pr, self.attr, 
97                            self.context.credentials())
98                elif name =='prover':
99                    self.me = self.chars.strip()
100                elif name =='principal':
101                    self.pr = self.chars.strip()
102                elif name == 'attribute':
103                    self.attr = self.chars.strip()
104                elif name == 'credential':
105                    c = b64decode(self.chars)
106                    if self.context.load_id_chunk(c) != ABAC.ABAC_CERT_SUCCESS:
107                        self.context.load_attribute_chunk(c)
108                else:
109                    print "Doh!"
110
111            def element_start(self, name, attrs):
112                self.chars = ''
113
114            def cdata(self, data):
115                self.chars += data
116
117
118            def __init__(self):
119                self.p = ParserCreate()
120                self.me = None
121                self.pr = None
122                self.attr = None
123                self.proof = None
124                self.context = ABAC.Context()
125                self.chars = ''
126
127                self.p.CharacterDataHandler = self.cdata
128                self.p.EndElementHandler = self.element_end
129                self.p.StartElementHandler = self.element_start
130
131        pp = parse_proof()
132        if filename: pp.p.ParseFile(open(filename, 'r'))
133        elif file: pp.p.ParseFile(file)
134        elif string: pp.p.Parse(string, True)
135        else: print "Doh1"
136
137        return pp.proof
138
139    @staticmethod
140    def from_dict(d):
141        """
142        Reverse the to_dict process
143        """
144        if d: 
145            if 'prover' in d and 'principal' in d and 'attribute' in d:
146                ac = ABAC.Context()
147                self = proof(d['prover'], d['principal'], d['attribute'], None)
148                for c in d.get('credential', []):
149                    if ac.load_id_chunk(c) != ABAC.ABAC_CERT_SUCCESS:
150                        ac.load_attribute_chunk(c)
151                self.creds = ac.credentials()
152                return self
153            else:
154                return None
155        else:
156            return None
Note: See TracBrowser for help on using the repository browser.