9 from snfOCCI.config import VOMS_CONFIG
13 SSL_CLIENT_S_DN_ENV = "SSL_CLIENT_S_DN"
14 SSL_CLIENT_CERT_ENV = "SSL_CLIENT_CERT"
15 SSL_CLIENT_CERT_CHAIN_0_ENV = "SSL_CLIENT_CERT_CHAIN_0"
18 class VomsError(exception.Error):
19 """Voms credential management error"""
23 1: ('nosocket', 'Socket problem'),
24 2: ('noident', 'Cannot identify itself (certificate problem)'),
25 3: ('comm', 'Server problem'),
26 4: ('param', 'Wrong parameters'),
27 5: ('noext', 'VOMS extension missing'),
28 6: ('noinit', 'Initialization error'),
29 7: ('time', 'Error in time checking'),
30 8: ('idcheck', 'User data in extension different from the real'),
31 9: ('extrainfo', 'VO name and URI missing'),
32 10: ('format', 'Wrong data format'),
33 11: ('nodata', 'Empty extension'),
34 12: ('parse', 'Parse error'),
35 13: ('dir', 'Directory error'),
36 14: ('sign', 'Signature error'),
37 15: ('server', 'Unidentifiable VOMS server'),
38 16: ('mem', 'Memory problems'),
39 17: ('verify', 'Generic verification error'),
40 18: ('type', 'Returned data of unknown type'),
41 19: ('order', 'Ordering different than required'),
42 20: ('servercode', 'Error from the server'),
43 21: ('notavail', 'Method not available'),
47 5: (400, "Bad Request"),
48 11: (400, "Bad Request"),
49 14: (401, "Not Authorized"),
53 def _get_cert_chain(ssl_info):
54 """Return certificate and chain from the ssl info in M2Crypto format"""
56 cert = ssl_info.get(SSL_CLIENT_CERT_ENV, "")
57 chain = ssl_info.get(SSL_CLIENT_CERT_CHAIN_0_ENV, "")
59 cert = M2Crypto.X509.load_cert_string(cert)
60 aux = M2Crypto.X509.load_cert_string(chain)
61 chain = M2Crypto.X509.X509_Stack()
67 def _get_voms_info(ssl_info):
68 """Extract voms info from ssl_info and return dict with it."""
71 cert, chain = _get_cert_chain(ssl_info)
72 except M2Crypto.X509.X509Error as e:
75 with voms_helper.VOMS(VOMS_CONFIG["vomsdir_path"],VOMS_CONFIG["ca_path"], VOMS_CONFIG["vomsapi_lib"]) as v:
77 voms_data = v.retrieve(cert, chain)
82 raise VomsError(v.error.value)
85 for attr in ('user', 'userca', 'server', 'serverca',
86 'voname', 'uri', 'version', 'serial',
87 ('not_before', 'date1'), ('not_after', 'date2')):
88 if isinstance(attr, basestring):
89 d[attr] = getattr(voms_data, attr)
91 d[attr[0]] = getattr(voms_data, attr[1])
94 for fqan in iter(voms_data.fqan):
97 d["fqans"].append(fqan)
102 def _split_fqan(fqan):
104 gets a fqan and returns a tuple containing
105 (vo/groups, role, capability)
108 capability = l.pop().split("=")[-1]
109 role = l.pop().split("=")[-1]
110 vogroup = "/".join(l)
111 return (vogroup, role, capability)
114 def _split_fqan(fqan):
116 gets a fqan and returns a tuple containing
117 (vo/groups, role, capability)
120 capability = l.pop().split("=")[-1]
121 role = l.pop().split("=")[-1]
122 vogroup = "/".join(l)
123 return (vogroup, role, capability)
126 def authenticate(ssl_data):
128 voms_info = _get_voms_info(ssl_data)
129 except VomsError as e:
132 user_dn = voms_info["user"]
133 user_vo = voms_info["voname"]
134 user_fqans = voms_info["fqans"]
136 return user_dn, user_vo, user_fqans