1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
|
# vim:fileencoding=utf8:et:ts=4:sts=4:sw=4:ft=python
from django.contrib.auth import get_user_model
from django.contrib.auth.backends import ModelBackend
from django.db import IntegrityError
from okupy.accounts.models import LDAPUser
from okupy.common.ldap_helpers import get_bound_ldapuser
from OpenSSL.crypto import load_certificate, FILETYPE_PEM
import ldap
import paramiko
import base64
class LDAPAuthBackend(ModelBackend):
"""
Authentication backend that authenticates against LDAP password.
If authentication succeeds, it sets up secondary password
for the session.
"""
def authenticate(self, request, username, password):
# LDAP is case- and whitespace-insensitive
# we do normalization to avoid duplicate django db entries
# and help mockldap
username = username.lower().strip()
try:
bound_ldapuser = get_bound_ldapuser(
request=request,
username=username,
password=password)
with bound_ldapuser as u:
UserModel = get_user_model()
attr_dict = {
UserModel.USERNAME_FIELD: u.username
}
user = UserModel(**attr_dict)
try:
user.save()
except IntegrityError:
user = UserModel.objects.get(**attr_dict)
return user
except ldap.INVALID_CREDENTIALS:
return None
except ldap.STRONG_AUTH_REQUIRED:
return None
class SSLCertAuthBackend(ModelBackend):
"""
Authentication backend taht uses client certificate information.
It requires one of owner e-mails to match in LDAP.
"""
def authenticate(self, request):
# it can be: SUCCESS, NONE and likely some string for failure ;)
cert_verify = request.META.get('SSL_CLIENT_VERIFY', None)
if cert_verify != 'SUCCESS':
return None
# curious enough, it's easier to parse the whole certificate
# than DN obtained from it by nginx...
cert = load_certificate(FILETYPE_PEM,
request.META['SSL_CLIENT_RAW_CERT'])
dn = cert.get_subject().get_components()
# for multiple addresses, there are multiple emailAddress fields
for k, v in dn:
if k == 'emailAddress':
try:
u = LDAPUser.objects.get(email__contains=v)
except LDAPUser.DoesNotExist:
pass
else:
UserModel = get_user_model()
attr_dict = {
UserModel.USERNAME_FIELD: u.username
}
user = UserModel(**attr_dict)
try:
user.save()
except IntegrityError:
user = UserModel.objects.get(**attr_dict)
return user
return None
class SSHKeyAuthBackend(ModelBackend):
"""
Authentication backend that uses SSH keys stored in LDAP.
"""
def authenticate(self, ssh_key=None):
for u in LDAPUser.objects.all():
for k in u.ssh_key:
spl = k.split()
if len(spl) < 2:
continue
form, user_key = spl[:2]
if form == 'ssh-rsa':
key_class = paramiko.RSAKey
elif form == 'ssh-dss':
key_class = paramiko.DSSKey
else:
# key format not supported
continue
try:
user_key = key_class(data=base64.b64decode(user_key))
except (TypeError, paramiko.SSHException):
continue
# paramiko reconstructs the key, so simple match should be fine
if ssh_key == user_key:
UserModel = get_user_model()
attr_dict = {
UserModel.USERNAME_FIELD: u.username
}
user = UserModel(**attr_dict)
try:
user.save()
except IntegrityError:
user = UserModel.objects.get(**attr_dict)
return user
return None
|
GitWeb