globuzma/canaille-globuzma
1
0
Fork 0
forked from Github-Mirrors/canaille
Code Pull requests Activity

python to ldap two-ways serialization

Browse source
This commit is contained in:
Éloi Rivard 2021-12-08 15:53:20 +01:00
parent 015d410fb6
commit 65dd61c524
8 changed files with 79 additions and 64 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
canaille/admin/clients.py
View file

@ -142,7 +142,7 @@ def add(user):
return render_template("admin/client_add.html", form=form, menuitem="admin")
client_id = gen_salt(24)
client_id_issued_at = datetime.datetime.now().strftime("%Y%m%d%H%M%SZ")
client_id_issued_at = datetime.datetime.now()
client = Client(
oauthClientID=client_id,
oauthIssueDate=client_id_issued_at,
@ -161,7 +161,7 @@ def add(user):
oauthSoftwareVersion=form["oauthSoftwareVersion"].data,
oauthJWK=form["oauthJWK"].data,
oauthJWKURI=form["oauthJWKURI"].data,
oauthPreconsent="TRUE" if form["oauthPreconsent"].data else "FALSE",
oauthPreconsent=form["oauthPreconsent"].data,
oauthClientSecret=""
if form["oauthTokenEndpointAuthMethod"].data == "none"
else gen_salt(48),
@ -225,7 +225,7 @@ def client_edit(client_id):
oauthJWK=form["oauthJWK"].data,
oauthJWKURI=form["oauthJWKURI"].data,
oauthAudience=form["oauthAudience"].data,
oauthPreconsent="TRUE" if form["oauthPreconsent"].data else "FALSE",
oauthPreconsent=form["oauthPreconsent"].data,
)
client.save()
flash(

49
canaille/ldaputils.py
View file

@ -1,5 +1,7 @@
import datetime
import ldap
import ldap.filter
import warnings
from flask import g
@ -153,20 +155,48 @@ class LDAPObject:
return cls._attribute_type_by_name
@staticmethod
def ldap_to_python(name, value):
syntax = LDAPObject.ldap_object_attributes()[name].syntax
if syntax == "1.3.6.1.4.1.1466.115.121.1.24": # Generalized Time
value = value.decode("utf-8")
return datetime.datetime.strptime(value, "%Y%m%d%H%M%SZ") if value else None
if syntax == "1.3.6.1.4.1.1466.115.121.1.27": # Integer
return int(value.decode("utf-8"))
if syntax == "1.3.6.1.4.1.1466.115.121.1.7": # Boolean
return value.decode("utf-8").upper() == "TRUE"
return value.decode("utf-8")
@staticmethod
def python_to_ldap(name, value):
syntax = LDAPObject.ldap_object_attributes()[name].syntax
if syntax == "1.3.6.1.4.1.1466.115.121.1.24": # Generalized Time
return value.strftime("%Y%m%d%H%M%SZ").encode("utf-8")
if syntax == "1.3.6.1.4.1.1466.115.121.1.27": # Integer
return str(value).encode("utf-8")
if syntax == "1.3.6.1.4.1.1466.115.121.1.7": # Boolean
return ("TRUE" if value else "FALSE").encode("utf-8")
return value.encode("utf-8")
@staticmethod
def ldap_attrs_to_python(attrs):
return {
name: [value.decode("utf-8") for value in values]
name: [LDAPObject.ldap_to_python(name, value) for value in values]
for name, values in attrs.items()
}
@staticmethod
def python_attrs_to_ldap(attrs):
return {
name: [
value.encode("utf-8") if isinstance(value, str) else value
for value in values
]
name: [LDAPObject.python_to_ldap(name, value) for value in values]
for name, values in attrs.items()
}
@ -190,9 +220,10 @@ class LDAPObject:
for name, value in self.changes.items()
if value and value[0] and self.attrs.get(name) != value
}
changes = self.python_attrs_to_ldap(changes)
formatted_changes = self.python_attrs_to_ldap(changes)
modlist = [
(ldap.MOD_REPLACE, name, values) for name, values in changes.items()
(ldap.MOD_REPLACE, name, values)
for name, values in formatted_changes.items()
]
conn.modify_s(self.dn, modlist)
@ -203,8 +234,8 @@ class LDAPObject:
for name, value in {**self.attrs, **self.changes}.items()
if value and value[0]
}
changes = self.python_attrs_to_ldap(changes)
attributes = [(name, values) for name, values in changes.items()]
formatted_changes = self.python_attrs_to_ldap(changes)
attributes = [(name, values) for name, values in formatted_changes.items()]
conn.add_s(self.dn, attributes)
self.attrs = {**self.attrs, **self.changes}

57
canaille/models.py
View file

@ -215,15 +215,11 @@ class Client(LDAPObject, ClientMixin):
@property
def issue_date(self):
return (
datetime.datetime.strptime(self.oauthIssueDate, "%Y%m%d%H%M%SZ")
if self.oauthIssueDate
else None
)
return self.oauthIssueDate
@property
def preconsent(self):
return self.oauthPreconsent and self.oauthPreconsent.lower() == "true"
return self.oauthPreconsent
def get_client_id(self):
return self.oauthClientID
@ -269,11 +265,7 @@ class AuthorizationCode(LDAPObject, AuthorizationCodeMixin):
@property
def issue_date(self):
return (
datetime.datetime.strptime(self.oauthIssueDate, "%Y%m%d%H%M%SZ")
if self.oauthIssueDate
else None
)
return self.oauthIssueDate
def get_redirect_uri(self):
return self.oauthRedirectURI
@ -286,16 +278,17 @@ class AuthorizationCode(LDAPObject, AuthorizationCodeMixin):
def is_expired(self):
return (
datetime.datetime.strptime(self.oauthAuthorizationDate, "%Y%m%d%H%M%SZ")
self.oauthAuthorizationDate
+ datetime.timedelta(seconds=int(self.oauthAuthorizationLifetime))
< datetime.datetime.now()
)
def get_auth_time(self):
auth_time = datetime.datetime.strptime(
self.oauthAuthorizationDate, "%Y%m%d%H%M%SZ"
return int(
(
self.oauthAuthorizationDate - datetime.datetime(1970, 1, 1)
).total_seconds()
)
return int((auth_time - datetime.datetime(1970, 1, 1)).total_seconds())
class Token(LDAPObject, TokenMixin):
@ -305,17 +298,13 @@ class Token(LDAPObject, TokenMixin):
@property
def issue_date(self):
return (
datetime.datetime.strptime(self.oauthIssueDate, "%Y%m%d%H%M%SZ")
if self.oauthIssueDate
else None
)
return self.oauthIssueDate
@property
def expire_date(self):
return datetime.datetime.strptime(
self.oauthIssueDate, "%Y%m%d%H%M%SZ"
) + datetime.timedelta(seconds=int(self.oauthTokenLifetime))
return self.oauthIssueDate + datetime.timedelta(
seconds=int(self.oauthTokenLifetime)
)
@property
def revoked(self):
@ -331,12 +320,14 @@ class Token(LDAPObject, TokenMixin):
return int(self.oauthTokenLifetime)
def get_issued_at(self):
issue_date = datetime.datetime.strptime(self.oauthIssueDate, "%Y%m%d%H%M%SZ")
return int((issue_date - datetime.datetime(1970, 1, 1)).total_seconds())
return int(
(self.oauthIssueDate - datetime.datetime(1970, 1, 1)).total_seconds()
)
def get_expires_at(self):
issue_date = datetime.datetime.strptime(self.oauthIssueDate, "%Y%m%d%H%M%SZ")
issue_timestamp = (issue_date - datetime.datetime(1970, 1, 1)).total_seconds()
issue_timestamp = (
self.oauthIssueDate - datetime.datetime(1970, 1, 1)
).total_seconds()
return int(issue_timestamp) + int(self.oauthTokenLifetime)
def is_refresh_token_active(self):
@ -347,7 +338,7 @@ class Token(LDAPObject, TokenMixin):
def is_expired(self):
return (
datetime.datetime.strptime(self.oauthIssueDate, "%Y%m%d%H%M%SZ")
self.oauthIssueDate
+ datetime.timedelta(seconds=int(self.oauthTokenLifetime))
< datetime.datetime.now()
)
@ -366,18 +357,14 @@ class Consent(LDAPObject):
@property
def issue_date(self):
return (
datetime.datetime.strptime(self.oauthIssueDate, "%Y%m%d%H%M%SZ")
if self.oauthIssueDate
else None
)
return self.oauthIssueDate
@property
def revokation_date(self):
return datetime.datetime.strptime(self.oauthRevokationDate, "%Y%m%d%H%M%SZ")
return self.oauthRevokationDate
def revoke(self):
self.oauthRevokationDate = datetime.datetime.now().strftime("%Y%m%d%H%M%SZ")
self.oauthRevokationDate = datetime.datetime.now()
self.save()
tokens = Token.filter(

2
canaille/oauth.py
View file

@ -136,7 +136,7 @@ def authorize():
oauthClient=client.dn,
oauthSubject=user.dn,
oauthScope=scopes,
oauthIssueDate=datetime.datetime.now().strftime("%Y%m%d%H%M%SZ"),
oauthIssueDate=datetime.datetime.now(),
)
consent.save()

10
canaille/oauth2utils.py
View file

@ -88,7 +88,7 @@ def save_authorization_code(code, request):
oauthRedirectURI=request.redirect_uri or request.client.oauthRedirectURIs[0],
oauthScope=request.scope,
oauthNonce=nonce,
oauthAuthorizationDate=now.strftime("%Y%m%d%H%M%SZ"),
oauthAuthorizationDate=now,
oauthAuthorizationLifetime=str(84000),
oauthCodeChallenge=request.data.get("code_challenge"),
oauthCodeChallengeMethod=request.data.get("code_challenge_method"),
@ -153,9 +153,7 @@ class RefreshTokenGrant(_RefreshTokenGrant):
return user.dn
def revoke_old_credential(self, credential):
credential.oauthRevokationDate = datetime.datetime.now().strftime(
"%Y%m%d%H%M%SZ"
)
credential.oauthRevokationDate = datetime.datetime.now()
credential.save()
@ -201,7 +199,7 @@ def save_token(token, request):
t = Token(
oauthTokenType=token["token_type"],
oauthAccessToken=token["access_token"],
oauthIssueDate=now.strftime("%Y%m%d%H%M%SZ"),
oauthIssueDate=now,
oauthTokenLifetime=str(token["expires_in"]),
oauthScope=token["scope"],
oauthClient=request.client.dn,
@ -241,7 +239,7 @@ class RevocationEndpoint(_RevocationEndpoint):
return None
def revoke_token(self, token):
token.oauthRevokationDate = datetime.datetime.now().strftime("%Y%m%d%H%M%SZ")
token.oauthRevokationDate = datetime.datetime.now()
token.save()

6
tests/commands/test_clean.py
View file

@ -16,7 +16,7 @@ def test_clean_command(testclient, slapd_connection, client, user):
oauthNonce="nonce",
oauthAuthorizationDate=(
datetime.datetime.now() - datetime.timedelta(days=1)
).strftime("%Y%m%d%H%M%SZ"),
),
oauthAuthorizationLifetime="3600",
oauthCodeChallenge="challenge",
oauthCodeChallengeMethod="method",
@ -32,9 +32,7 @@ def test_clean_command(testclient, slapd_connection, client, user):
oauthTokenType=None,
oauthRefreshToken=gen_salt(48),
oauthScope="openid profile",
oauthIssueDate=(datetime.datetime.now() - datetime.timedelta(days=1)).strftime(
"%Y%m%d%H%M%SZ"
),
oauthIssueDate=(datetime.datetime.now() - datetime.timedelta(days=1)),
oauthTokenLifetime=str(3600),
)
token.save(slapd_connection)

11
tests/conftest.py
View file

@ -234,7 +234,7 @@ def client(app, slapd_connection, other_client):
"https://mydomain.tld/redirect2",
],
oauthLogoURI="https://mydomain.tld/logo.png",
oauthIssueDate=datetime.datetime.now().strftime("%Y%m%d%H%S%MZ"),
oauthIssueDate=datetime.datetime.now(),
oauthClientSecret=gen_salt(48),
oauthGrantType=[
"password",
@ -268,7 +268,7 @@ def other_client(app, slapd_connection):
"https://myotherdomain.tld/redirect2",
],
oauthLogoURI="https://myotherdomain.tld/logo.png",
oauthIssueDate=datetime.datetime.now().strftime("%Y%m%d%H%S%MZ"),
oauthIssueDate=datetime.datetime.now(),
oauthClientSecret=gen_salt(48),
oauthGrantType=[
"password",
@ -300,7 +300,7 @@ def authorization(app, slapd_connection, user, client):
oauthResponseType="code",
oauthScope="openid profile",
oauthNonce="nonce",
oauthAuthorizationDate="20200101000000Z",
oauthAuthorizationDate=datetime.datetime(2020, 1, 1),
oauthAuthorizationLifetime="3600",
oauthCodeChallenge="challenge",
oauthCodeChallengeMethod="method",
@ -313,6 +313,7 @@ def authorization(app, slapd_connection, user, client):
@pytest.fixture
def user(app, slapd_connection):
User.ldap_object_classes(slapd_connection)
LDAPObject.ldap_object_attributes(slapd_connection)
u = User(
objectClass=["inetOrgPerson"],
cn="John (johnny) Doe",
@ -365,7 +366,7 @@ def token(slapd_connection, client, user):
oauthTokenType=None,
oauthRefreshToken=gen_salt(48),
oauthScope="openid profile",
oauthIssueDate=datetime.datetime.now().strftime("%Y%m%d%H%M%SZ"),
oauthIssueDate=datetime.datetime.now(),
oauthTokenLifetime=str(3600),
)
t.save(slapd_connection)
@ -378,7 +379,7 @@ def consent(slapd_connection, client, user):
oauthClient=client.dn,
oauthSubject=user.dn,
oauthScope=["openid", "profile"],
oauthIssueDate=datetime.datetime.now().strftime("%Y%m%d%H%M%SZ"),
oauthIssueDate=datetime.datetime.now(),
)
t.save(slapd_connection)
return t

2
tests/test_authorization_code_flow.py
View file

@ -67,7 +67,7 @@ def test_authorization_code_flow(
def test_authorization_code_flow_preconsented(
testclient, slapd_connection, logged_user, client, keypair, other_client
):
client.oauthPreconsent = "TRUE"
client.oauthPreconsent = True
client.save(conn=slapd_connection)
res = testclient.get(