unicef-security

Pipfile

@@ -77,6 +77,7 @@ unicef-djangolib = "==0.5.4"
 unicef-locations = "==4.0.1"
 unicef-notification = "==1.1"
 unicef-restlib = "==0.7"
+unicef-security = "==1.1"
 unicef-snapshot = "==1.2"
 unicef-rest-export = "==0.6"
 xhtml2pdf = "==0.2.7"

src/etools/applications/core/auth.py

@@ -3,7 +3,6 @@
 from django.conf import settings
 from django.contrib.auth import get_user_model
 from django.contrib.auth.models import Group
-from django.http import HttpResponseRedirect
 
 from rest_framework.authentication import (
     BasicAuthentication,
@@ -13,59 +12,14 @@
 )
 from rest_framework.exceptions import PermissionDenied
 from rest_framework_simplejwt.authentication import JWTAuthentication
-from social_core.backends.azuread_b2c import AzureADB2COAuth2
-from social_core.exceptions import AuthCanceled, AuthMissingParameter
-from social_core.pipeline import social_auth, user as social_core_user
-from social_django.middleware import SocialAuthExceptionMiddleware
+from social_core.pipeline import user as social_core_user
 
 from etools.applications.users.models import Country
 from etools.libraries.tenant_support.utils import set_country
 
 logger = logging.getLogger(__name__)
 
 
-def social_details(backend, details, response, *args, **kwargs):
-    r = social_auth.social_details(backend, details, response, *args, **kwargs)
-    r['details']['idp'] = response.get('idp')
-
-    if not r['details'].get('email'):
-        if not response.get('email'):
-            r['details']['email'] = response["signInNames.emailAddress"]
-        else:
-            r['details']['email'] = response.get('email')
-
-    email = r['details'].get('email')
-    if isinstance(email, str):
-        r['details']['email'] = email.lower()
-    return r
-
-
-def get_username(strategy, details, backend, user=None, *args, **kwargs):
-    return {'username': details.get('email')}
-
-
-def create_user(strategy, details, backend, user=None, *args, **kwargs):
-    """ Overwrite social_account.user.create_user to only create new users if they're UNICEF"""
-
-    if user:
-        return {'is_new': False}
-
-    fields = dict((name, kwargs.get(name, details.get(name)))
-                  for name in backend.setting('USER_FIELDS', social_core_user.USER_FIELDS))
-    if not fields:
-        return
-
-    response = kwargs.get('response')
-    if response:
-        email = response.get('email') or response.get("signInNames.emailAddress")
-        if not email.endswith("unicef.org"):
-            return
-    return {
-        'is_new': True,
-        'user': strategy.create_user(**fields)
-    }
-
-
 def user_details(strategy, details, backend, user=None, *args, **kwargs):
     # This is where we update the user
     # see what the property to map by is here
@@ -96,47 +50,6 @@ def user_details(strategy, details, backend, user=None, *args, **kwargs):
     return social_core_user.user_details(strategy, details, backend, user, *args, **kwargs)
 
 
-class CustomAzureADBBCOAuth2(AzureADB2COAuth2):
-    BASE_URL = 'https://{tenant_id}.b2clogin.com/{tenant_id}.onmicrosoft.com'
-
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.redirect_uri = settings.HOST + '/social/complete/azuread-b2c-oauth2/'
-
-
-class CustomSocialAuthExceptionMiddleware(SocialAuthExceptionMiddleware):
-
-    def process_exception(self, request, exception):
-        if isinstance(exception, (AuthCanceled, AuthMissingParameter)):
-            return HttpResponseRedirect(self.get_redirect_uri(request, exception))
-        else:
-            raise exception
-
-    def get_redirect_uri(self, request, exception):
-        error = request.GET.get('error', None)
-
-        # This is what we should expect:
-        # ['AADB2C90118: The user has forgotten their password.\r\n
-        # Correlation ID: 7e8c3cf9-2fa7-47c7-8924-a1ea91137ba9\r\n
-        # Timestamp: 2018-11-13 11:37:56Z\r\n']
-        error_description = request.GET.get('error_description', None)
-
-        if error == "access_denied" and error_description is not None:
-            if 'AADB2C90118' in error_description:
-                auth_class = CustomAzureADBBCOAuth2()
-                redirect_home = auth_class.get_redirect_uri()
-                redirect_url = auth_class.base_url + '/oauth2/v2.0/' + \
-                    'authorize?p=' + settings.SOCIAL_PASSWORD_RESET_POLICY + \
-                    '&client_id=' + settings.KEY + \
-                    '&nonce=defaultNonce&redirect_uri=' + redirect_home + \
-                    '&scope=openid+email&response_type=code'
-
-                return redirect_url
-
-        # TODO: In case of password reset the state can't be verified figure out a way to log the user in after reset
-        return settings.LOGIN_URL
-
-
 class DRFBasicAuthMixin(BasicAuthentication):
     def authenticate(self, request):
         super_return = super().authenticate(request)

src/etools/applications/core/tests/test_auth.py

@@ -3,12 +3,15 @@
 from django.contrib.auth import get_user_model
 from django.test import SimpleTestCase
 
+from unicef_security import pipeline
+
 from etools.applications.core import auth
 from etools.applications.core.tests.cases import BaseTenantTestCase
 from etools.applications.users.tests.factories import UserFactory
 
-SOCIAL_AUTH_PATH = "etools.applications.core.auth.social_auth"
-SOCIAL_USER_PATH = "etools.applications.core.auth.social_core_user"
+SOCIAL_AUTH_PATH = "unicef_security.pipeline.social_auth"
+SOCIAL_USER_PATH = "unicef_security.pipeline.social_core_user"
+SOCIAL_ETOOLS_USER_PATH = "etools.applications.core.auth.social_core_user"
 
 
 class TestSocialDetails(SimpleTestCase):
@@ -27,7 +30,7 @@ def test_details_missing_email(self):
             'details': self.details
         }
         with patch(SOCIAL_AUTH_PATH, self.mock_social):
-            r = auth.social_details(
+            r = pipeline.social_details(
                 None,
                 {},
                 {"idp": "123", "email": "test@example.com"}
@@ -42,7 +45,7 @@ def test_details(self):
             'details': self.details
         }
         with patch(SOCIAL_AUTH_PATH, self.mock_social):
-            r = auth.social_details(
+            r = pipeline.social_details(
                 None,
                 {},
                 {"idp": "123", "email": "new@example.com"}
@@ -63,7 +66,7 @@ def setUp(self):
 
     def test_user_exists(self):
         self.user = UserFactory(username=self.details["email"])
-        r = auth.get_username(None, self.details, None)
+        r = pipeline.get_username(None, self.details, None)
         self.assertEqual(r, {"username": self.details["email"]})
 
 
@@ -81,7 +84,7 @@ def setUp(self):
 
     def test_no_user(self):
         with patch(SOCIAL_USER_PATH, self.mock_social):
-            r = auth.user_details("strategy", self.details, None, None)
+            r = pipeline.user_details("strategy", self.details, None, None)
         self.assertEqual(r, "Returned")
         self.mock_social.user_details.assert_called_with(
             "strategy",
@@ -97,7 +100,7 @@ def test_no_update(self):
         )
         self.details["business_area_code"] = user.profile.country.business_area_code
         with patch(SOCIAL_USER_PATH, self.mock_social):
-            r = auth.user_details("strategy", self.details, None, user)
+            r = pipeline.user_details("strategy", self.details, None, user)
         self.assertEqual(r, "Returned")
         self.mock_social.user_details.assert_called_with(
             "strategy",
@@ -116,7 +119,7 @@ def test_no_profile_country(self):
         user.profile.country = None
         user.profile.save()
         self.assertIsNone(user.profile.country)
-        with patch(SOCIAL_USER_PATH, self.mock_social):
+        with patch(SOCIAL_ETOOLS_USER_PATH, self.mock_social):
             r = auth.user_details("strategy", self.details, None, user)
         self.assertEqual(r, "Returned")
         self.mock_social.user_details.assert_called_with(
@@ -139,7 +142,7 @@ def test_is_staff_update(self):
         self.details["business_area_code"] = country.business_area_code
         self.details["idp"] = "UNICEF Azure AD"
         self.assertFalse(user.is_staff)
-        with patch(SOCIAL_USER_PATH, self.mock_social):
+        with patch(SOCIAL_ETOOLS_USER_PATH, self.mock_social):
             r = auth.user_details("strategy", self.details, None, user)
         self.assertEqual(r, "Returned")
         self.mock_social.user_details.assert_called_with(

src/etools/config/settings/base.py

@@ -118,7 +118,7 @@ def get_from_secrets_or_env(var_name, default=None):
 # DJANGO: HTTP
 MIDDLEWARE = (
     'django.contrib.sessions.middleware.SessionMiddleware',
-    'etools.applications.core.auth.CustomSocialAuthExceptionMiddleware',
+    'unicef_security.middleware.UNICEFSocialAuthExceptionMiddleware',
     'django.contrib.auth.middleware.AuthenticationMiddleware',
     'django.middleware.common.CommonMiddleware',
     'django.middleware.csrf.CsrfViewMiddleware',
@@ -196,6 +196,7 @@ def get_from_secrets_or_env(var_name, default=None):
     'etools.applications.tpm.tpmpartners',
     'waffle',
     'etools.applications.permissions2',
+    'unicef_security',
     'unicef_notification',
     'etools_offline',
     'etools.applications.offline',
@@ -282,7 +283,7 @@ def get_from_secrets_or_env(var_name, default=None):
 # CONTRIB: AUTH
 AUTHENTICATION_BACKENDS = (
     # 'social_core.backends.azuread_b2c.AzureADB2COAuth2',
-    'etools.applications.core.auth.CustomAzureADBBCOAuth2',
+    'unicef_security.backends.UNICEFAzureADB2COAuth2',
     'django.contrib.auth.backends.ModelBackend',
 )
 AUTH_USER_MODEL = 'users.User'
@@ -509,6 +510,7 @@ def before_send(event, hint):
 SOCIAL_AUTH_JSONFIELD_ENABLED = True
 POLICY = os.getenv('AZURE_B2C_POLICY_NAME', "b2c_1A_UNICEF_PARTNERS_signup_signin")
 
+TENANT_NAME = os.getenv('AZURE_B2C_TENANT_NAME', 'unicefpartners')
 TENANT_ID = os.getenv('AZURE_B2C_TENANT', 'unicefpartners')
 SCOPE = ['openid', 'email']
 IGNORE_DEFAULT_SCOPE = True
@@ -521,15 +523,15 @@ def before_send(event, hint):
 SOCIAL_PASSWORD_RESET_POLICY = os.getenv('AZURE_B2C_PASS_RESET_POLICY', "B2C_1_PasswordResetPolicy")
 SOCIAL_AUTH_PIPELINE = (
     # 'social_core.pipeline.social_auth.social_details',
-    'etools.applications.core.auth.social_details',
+    'unicef_security.pipeline.social_details',
     'social_core.pipeline.social_auth.social_uid',
     # allows based on emails being listed in 'WHITELISTED_EMAILS' or 'WHITELISTED_DOMAINS'
     'social_core.pipeline.social_auth.auth_allowed',
     'social_core.pipeline.social_auth.social_user',
     # 'social_core.pipeline.user.get_username',
-    'etools.applications.core.auth.get_username',
+    'unicef_security.pipeline.get_username',
     'social_core.pipeline.social_auth.associate_by_email',
-    'etools.applications.core.auth.create_user',
+    'unicef_security.pipeline.create_user',
     'social_core.pipeline.social_auth.associate_user',
     'social_core.pipeline.social_auth.load_extra_data',
     'social_core.pipeline.user.user_details',

src/etools/config/settings/local.py

@@ -25,7 +25,7 @@
 
 AUTHENTICATION_BACKENDS = (
     # 'social_core.backends.azuread_b2c.AzureADB2COAuth2',
-    'etools.applications.core.auth.CustomAzureADBBCOAuth2',
+    'unicef_security.backends.UNICEFAzureADB2COAuth2',
     'django.contrib.auth.backends.ModelBackend',
 )
 

src/etools/config/urls.py

@@ -98,6 +98,7 @@
     re_path(r'^api/v2/activity/', include('unicef_snapshot.urls')),
     re_path(r'^api/v2/environment/', include('etools.applications.environment.urls_v2')),
     re_path(r'^api/v2/attachments/', include('unicef_attachments.urls')),
+    re_path(r'^security/', include('unicef_security.urls')),
 
     # ***************  API version 3  ******************
     re_path(r'^api/v3/users/', include('etools.applications.users.urls_v3', namespace='users_v3')),