ajout système sync django -> Nexcloud pour les utilisateurs

nextcloud/README.rst

@@ -0,0 +1 @@
+../doc/nextcloud_app.rst

nextcloud/__init__.py

@@ -0,0 +1 @@
+default_app_config = 'nextcloud.apps.NextcloudConfig'

nextcloud/apps.py

@@ -0,0 +1,13 @@
+# -*- coding: utf-8 -*-
+from __future__ import unicode_literals
+
+from django.apps import AppConfig
+
+
+class NextcloudConfig(AppConfig):
+    name = 'nextcloud'
+
+    def ready(self):
+
+        # Load and connect signal recievers
+        import nextcloud.signals

nextcloud/models/__init__.py

@@ -0,0 +1,4 @@
+from nextcloud.models.alerts import *
+
+# User too for good measure
+from django.contrib.auth.models import User

nextcloud/models/alerts.py

@@ -0,0 +1,29 @@
+import logging
+
+from django.dispatch import Signal
+
+logger = logging.getLogger(__name__)
+
+new_membership = Signal(Signal("user"))
+ending_membership = Signal(Signal("user"))
+updating_membership = Signal(["user"])
+
+from django.db.models.signals import post_save, post_delete
+from django.contrib.auth.models import User
+
+def my_callback(sender,created,instance, **kwargs):
+    if created:
+        new_membership.send(sender=sender, user=instance)
+    else:
+        updating_membership.send(sender=sender, user=instance)
+
+
+def del_callback(sender, instance,  **kwargs):
+    ending_membership.send(sender=sender, user=instance)
+
+
+
+
+post_save.connect(my_callback, sender=User)
+post_delete.connect(del_callback, sender=User)
+

nextcloud/requirements.txt

@@ -0,0 +1,3 @@
+requests
+Django
+pyocclient

nextcloud/settings.py

@@ -0,0 +1,14 @@
+from django.conf import settings
+
+from .models.alerts import new_membership, ending_membership, updating_membership
+
+NEXTCLOUD_HOST = getattr(settings, 'NEXTCLOUD_HOST', 'cloud.ambulance-clerc.ch')
+NEXTCLOUD_ADMIN = getattr(settings, 'NEXTCLOUD_ADMIN', 'admin')
+NEXTCLOUD_PASSWORD = getattr(settings, 'NEXTCLOUD_SECRET', 'Mc33*144')
+
+NEXTCLOUD_USE_HTTPS = getattr(settings, 'NEXTCLOUD_USE_HTTPS', True)
+NEXTCLOUD_SSL_IS_SIGNED = getattr(settings, 'NEXTCLOUD_SSL_IS_SIGNED', True)
+
+NEXTCLOUD_USER_GROUP = getattr(settings, 'NEXTCLOUD_USER_GROUP', "autoGenerated")
+NEXTCLOUD_USER_QUOTA = getattr(settings, 'NEXTCLOUD_USER_QUOTA', '1GB')
+NEXTCLOUD_USER_DEFAULT_PASSWORD = getattr(settings, 'NEXTCLOUD_USER_DEFAULT_PASSWORD', 'Mc144*1870')

nextcloud/signals.py

@@ -0,0 +1,137 @@
+import logging
+import random
+import string
+import requests
+
+import owncloud as NextCloudCLI
+
+
+from django.dispatch import receiver
+
+from .settings import *
+
+logger = logging.getLogger(__name__)
+
+def log(msg):
+    logger.debug(msg)
+    print(msg)
+
+
+
+def randomStringDigits(stringLength=8):
+    # Generate a random string of letters and digits
+    lettersAndDigits = string.ascii_letters + string.digits
+    return ''.join(random.choice(lettersAndDigits) for i in range(stringLength))
+
+
+@receiver(new_membership)
+def adding_nexcloud_user(sender, **kwargs):
+    print("pass add nextcloud user")
+    # Add new member on the nextcloud server
+    _user = kwargs['user']
+    _pass = NEXTCLOUD_USER_DEFAULT_PASSWORD
+
+    _admin = NEXTCLOUD_ADMIN
+    _admin_key = NEXTCLOUD_PASSWORD
+
+    _url = '%s://%s' % (('http', 'https')[NEXTCLOUD_USE_HTTPS is True], NEXTCLOUD_HOST)
+
+    nc = NextCloudCLI.Client(_url, verify_certs=NEXTCLOUD_SSL_IS_SIGNED)
+    _err = ['login', 'create_user', 'enable_user', 'add_user_to_group', 'set_user_attribute quota']
+    try:
+        nc.login(_admin, _admin_key)
+        print("pass logged")
+        _err.pop(0)
+        try:
+            print(f"NC User Created  {_user.username} / {_pass}")
+            nc.create_user(_user.username, _pass)
+            print(f"**NC User Created passed {_user.username} / {_pass}")
+            _err.pop(0)
+        except NextCloudCLI.OCSResponseError as e:
+            if e.status_code == 102:
+                _err.pop(0)
+                nc.make_ocs_request('PUT', 'cloud/users', _user.username + '/enable')
+            else:
+                raise e
+        _err.pop(0)
+        if NEXTCLOUD_USER_GROUP is not None:
+            nc.add_user_to_group(_user.username, NEXTCLOUD_USER_GROUP)
+        _err.pop(0)
+        if NEXTCLOUD_USER_QUOTA is not None:
+            nc.set_user_attribute(_user.username, 'quota', NEXTCLOUD_USER_QUOTA)
+        nc.logout()
+
+    except NextCloudCLI.HTTPResponseError as e:
+        log('adding_nexcloug_user: %s HTTP Error %d' % (_err[0], e.status_code))
+    except requests.exceptions.ConnectionError as e:
+        log('adding_nexcloug_user: %s Connection Error: %s' % (_err[0], e))
+    except NextCloudCLI.OCSResponseError as e:
+        log('adding_nexcloug_user: %s OCSResponse Error %d' % (_err[0], e.status_code))
+    else:
+        log('Adding user %s success' % _user.username)
+
+
+@receiver(ending_membership)
+def deactivating_nexcloud_user(sender, **kwargs):
+    # Remove member on the nextcloud server
+    _user = kwargs['user']
+
+    _admin = NEXTCLOUD_ADMIN
+    _admin_key = NEXTCLOUD_PASSWORD
+
+    _url = ('http', 'https')[NEXTCLOUD_USE_HTTPS is True] + '://' + NEXTCLOUD_HOST
+
+    nc = NextCloudCLI.Client(_url, verify_certs=NEXTCLOUD_SSL_IS_SIGNED)
+    _err = ['login', 'deactivate_user']
+    try:
+        nc.login(_admin, _admin_key)
+        _err.pop(0)
+        nc.make_ocs_request('PUT', 'cloud/users', _user.username + '/disable')
+        nc.logout()
+
+    except NextCloudCLI.HTTPResponseError as e:
+        log('removing_nexcloug_user: %s HTTP Error %d' % (_err[0], e.status_code))
+    except requests.exceptions.ConnectionError as e:
+        log('removing_nexcloug_user: %s Connection Error: %s' % (_err[0], e))
+    except NextCloudCLI.OCSResponseError as e:
+        log('removing_nexcloug_user: %s OCSResponse Error %d' % (_err[0], e.status_code))
+    else:
+        log('Removing user %s success' % _user.username)
+        print('Removing user %s success' % _user.username)
+
+
+
+@receiver(updating_membership)
+def update_nexcloud_user(sender, **kwargs):
+    # Change user password on the nextcloud server
+    _user = kwargs['user']
+
+    _admin = NEXTCLOUD_ADMIN
+    _admin_key = NEXTCLOUD_PASSWORD
+
+    _url = ('http', 'https')[NEXTCLOUD_USE_HTTPS is True] + '://' + NEXTCLOUD_HOST
+
+    nc = NextCloudCLI.Client(_url, verify_certs=NEXTCLOUD_SSL_IS_SIGNED)
+    _err = ['login', 'set_user_attribute password']
+    try:
+        nc.login(_admin, _admin_key)
+        _err.pop(0)
+        nc.set_user_attribute(_user.username, 'email', _user.email)
+        nc.set_user_attribute(_user.username, 'displayname', _user.first_name + " " + _user.last_name + get_pseudonyme_from_user(_user.username))
+        nc.logout()
+
+    except NextCloudCLI.HTTPResponseError as e:
+        log('changing_nexcloud_user_password: %s HTTP Error %d' % (_err[0], e.status_code))
+    except requests.exceptions.ConnectionError as e:
+        log('changing_nexcloud_user_password: %s Connection Error: %s' % (_err[0], e))
+    except NextCloudCLI.OCSResponseError as e:
+        log('changing_nexcloud_user_password: %s OCSResponse Error %d' % (_err[0], e.status_code))
+    else:
+        log('Update user %s success' % _user.username)
+
+def get_pseudonyme_from_user(username):
+    print(username)
+    if "mc" in username:
+        return " (" + username.split("@")[0].replace("mc","MC") + ")"
+    else:
+        return ""

nextcloud/tests.py

@@ -0,0 +1,43 @@
+from django.test import TestCase
+from django.contrib.auth.models import User
+
+from unittest.mock import patch
+
+from .models.alerts import MemberAlertManager
+from .signals import randomStringDigits
+
+import logging
+
+logging.basicConfig(level=logging.DEBUG)
+
+_logger = logging.getLogger(__name__)
+
+
+class NextcloudTestCase(TestCase):
+
+    myClass = MemberAlertManager()
+    _pass = randomStringDigits()
+
+    def setUp(self):
+        self._usr = User.objects.create_user('4321_test_1234', email='test@example.com', password='password')
+
+    def tearDown(self):
+        self._usr.delete()
+
+    @patch('nextcloud.signals.logger')
+    def test_adding_nexcloud_user(self, mock_logger):
+        self.myClass.handle_new_membership(self._usr)
+        mock_logger.debug.assert_called_with('Adding user %s success' % self._usr.username)
+        _logger.debug('test_adding_nexcloud_user %s success' % self._usr.username)
+
+    @patch('nextcloud.signals.logger')
+    def test_changing_nexcloud_user_password(self, mock_logger):
+        self.myClass.handle_changing_membership_password(self._usr, self._pass)
+        mock_logger.debug.assert_called_with('Changing user %s password success' % self._usr.username)
+        _logger.debug('test_changing_nexcloud_user_password %s success' % self._usr)
+
+    @patch('nextcloud.signals.logger')
+    def test_removing_nexcloud_user(self, mock_logger):
+        self.myClass.handle_ending_membership(self._usr)
+        mock_logger.debug.assert_called_with('Removing user %s success' % self._usr.username)
+        _logger.debug('test_removing_nexcloud_user %s success' % self._usr)