Latest changes

djeveric/__init__.py

@@ -1,154 +0,0 @@
-from django.contrib.auth import get_user_model
-from django.contrib.auth.tokens import default_token_generator
-from django.contrib.contenttypes.models import ContentType
-from django.utils.encoding import force_bytes
-from django.utils.http import urlsafe_base64_encode, urlsafe_base64_decode
-from rest_framework import serializers, status
-from rest_framework.exceptions import ValidationError
-from rest_framework.response import Response
-from rest_framework.settings import api_settings
-from rest_framework.views import APIView
-
-
-def encode_pk(resource):
-    """Encode the primary key of a resource with Base64 for usage in URLs."""
-    return urlsafe_base64_encode(force_bytes(resource.pk))
-
-
-class ConfirmationError(Exception):
-    """An error occurring while checking a confirmation."""
-
-    def __init__(self, message=None, errors=None):
-        if errors is None:
-            self.errors = {}
-        else:
-            self.errors = errors
-        if message is not None:
-            non_field_errors = self.errors.get(api_settings.NON_FIELD_ERRORS_KEY, [])
-            non_field_errors.append(message)
-            self.errors[api_settings.NON_FIELD_ERRORS_KEY] = non_field_errors
-
-
-class Email:
-    """
-    Base class for an email message.
-    """
-    subject = ""
-
-    def __init__(self, user):
-        self.user = user
-
-    def get_subject(self):
-        return self.subject
-
-    def get_message(self, context):
-        raise NotImplementedError()
-
-    def send(self, context):
-        self.user.email_user(self.get_subject(), self.get_message(context))
-
-
-class Confirmation:
-    """
-    Base class for handling a confirmation process.
-    """
-    email_class = Email
-
-    def __init__(self, user, resource):
-        self.user = user
-        self.resource = resource
-
-    def check(self):
-        if not self.has_permission(self.user, self.resource):
-            raise ConfirmationError("Permission denied")
-        if self.is_confirmed(self.resource):
-            raise ConfirmationError("Already confirmed")
-        self.confirm(self.resource)
-
-    def confirm(self, resource):
-        """Overwrite this method to supply operations to confirm the resource."""
-        pass
-
-    def get_email(self):
-        return self.email_class(self.user)
-
-    def has_permission(self, user, resource) -> bool:
-        """Overwrite this method returning if a user may confirm a resource."""
-        return False
-
-    def is_confirmed(self, resource) -> bool:
-        """Overwrite this method to tell if a resource is confirmed."""
-        return False
-
-    def send_request(self):
-        if self.is_confirmed(self.resource):
-            return
-        self.get_email().send({
-            "token": default_token_generator.make_token(self.user),
-            "uid": encode_pk(self.user),
-            "rtid": encode_pk(ContentType.objects.get_for_model(self.resource)),
-            "rid": encode_pk(self.resource),
-        })
-
-
-class ConfirmationSerializer(serializers.Serializer):
-    """
-    Serializer class for confirmation requests.
-    """
-    token = serializers.CharField()
-    uid = serializers.CharField()
-    rtid = serializers.CharField()
-    rid = serializers.CharField()
-
-    def __init__(self, **kwargs):
-        super().__init__(**kwargs)
-        self.user = None
-        self.resource_type = None
-        self.resource = None
-
-    def validate_uid(self, value):
-        self.user = self.get_model_object(get_user_model(), value)
-        return value
-
-    def validate_rtid(self, value):
-        self.resource_type = self.get_model_object(ContentType, value)
-        return value
-
-    def validate(self, data):
-        # we need to be sure that the rtid was already decoded
-        self.resource = self.get_model_object(self.resource_type.model_class(), data["rid"])
-        return data
-
-    def get_model_object(self, model, oid):
-        try:
-            # urlsafe_base64_decode() decodes to bytestring
-            oid = urlsafe_base64_decode(oid).decode()
-            return model._default_manager.get(id=oid)
-        except (TypeError, ValueError, OverflowError, model.DoesNotExist):
-            raise ValidationError("Error while decoding object id")
-
-
-class ConfirmationView(APIView):
-    """
-    View for creating a confirmation API endpoint.
-    """
-    confirmation_class = Confirmation
-    serializer_class = ConfirmationSerializer
-
-    def post(self, request, format=None):
-        try:
-            self.check_confirmation(request.data)
-            return Response({}, status=status.HTTP_204_NO_CONTENT)
-        except ConfirmationError as e:
-            return Response(e.errors, status=status.HTTP_400_BAD_REQUEST)
-
-    def check_confirmation(self, data):
-        serializer = self.serializer_class(data=data)
-        if not serializer.is_valid():
-            raise ConfirmationError(errors=serializer.errors)
-        token = serializer.validated_data["token"]
-        if default_token_generator.check_token(serializer.user, token):
-            # confirm resource or raise an exception otherwise
-            self.confirmation_class(serializer.user, serializer.resource).check()
-        else:
-            raise ConfirmationError("Invalid token")

requirements.txt

@@ -1,3 +1,4 @@
 setuptools~=40.8.0
 django~=2.2.13
-djangorestframework~=3.9.0
+djangorestframework~=3.9.0
+ldap3~=2.4.1

userausfall/ldap.py

@@ -3,6 +3,36 @@ from ldap3 import Server, Connection, SYNC
 
 
 def create_account(username, raw_password):
+    connection = _get_connection()
+    is_success = connection.add(
+        f"cn={username},dc=local",
+        ["simpleSecurityObject", "organizationalRole"],
+        {"userPassword": raw_password},
+    )
+    return is_success
+
+
+def account_exists(username):
+    connection = _get_connection()
+    exists = connection.search(
+        f"cn={username},dc=local", "(objectclass=simpleSecurityObject)"
+    )
+    return exists
+
+
+def is_valid_account_data(username, raw_password):
+    connection = _get_connection()
+    is_valid = connection.search(
+        f"cn={username},dc=local",
+        "(objectclass=simpleSecurityObject)",
+        attributes=["userPassword"],
+    )
+    if is_valid:
+        is_valid = connection.entries[0]["userPassword"].value == raw_password
+    return is_valid
+
+
+def _get_connection():
     server = Server("localhost")
     # The SAFE_SYNC client strategy doesn't seem to be present in Buster version of ldap3. We might want to use it as
     # soon as it is available (multithreading).
@@ -13,9 +43,4 @@ def create_account(username, raw_password):
         client_strategy=SYNC,
         auto_bind=True,
     )
-    is_success = connection.add(
+    return connection
-        f"cn={username},dc=local",
-        ["simpleSecurityObject", "organizationalRole"],
-        {"userPassword": raw_password},
-    )
-    return is_success