1 | #/usr/local/bin/python |
---|
2 | |
---|
3 | |
---|
4 | import ABAC |
---|
5 | import sys |
---|
6 | import os, os.path |
---|
7 | |
---|
8 | from string import join |
---|
9 | from base64 import b64encode, b64decode |
---|
10 | from xml.parsers.expat import ParserCreate |
---|
11 | |
---|
12 | class 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 |
---|