Skip to content

sso_client#

SSOClient#

Client for synchronizing user data using Fairdata SSO.

Source code in src/apps/users/sso_client.py 
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
class SSOClient:
    """Client for synchronizing user data using Fairdata SSO."""

    def __init__(self):
        self.enabled = settings.ENABLE_SSO_AUTH
        self.host = settings.SSO_HOST
        self.trusted_service_token = settings.SSO_TRUSTED_SERVICE_TOKEN
        if self.enabled and not (self.trusted_service_token and self.host):
            self.enabled = False
            _logger.warning(
                "User sync disabled due to missing SSO_TRUSTED_SERVICE_TOKEN or SSO_HOST."
            )

    def sync_user(self, user: MetaxUser) -> bool:
        """Sync user from SSO if necessary."""
        if not self.enabled:
            return False

        if not getattr(user, "fairdata_username", None):
            return False  # Non-fairdata user, no need to sync

        # No need to sync active user if synced recently
        now = timezone.now()
        if user.is_active and user.synced and (now - user.synced < timezone.timedelta(hours=8)):
            return False

        payload = {
            "id": user.username,
            "token": self.trusted_service_token,
        }
        url = f"{self.host}/user_status"
        res = requests.post(url, payload)
        if res.status_code != 200:
            _logger.warning("User sync error. Failed to get user data from {url}: {res.text} ")
            return False

        # Update user
        data = res.json()
        user.synced = now
        user.is_active = not data["locked"]
        user.csc_projects = data["projects"]
        user.save()
        return True

sync_user(user) #

Sync user from SSO if necessary.

Source code in src/apps/users/sso_client.py 
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
def sync_user(self, user: MetaxUser) -> bool:
    """Sync user from SSO if necessary."""
    if not self.enabled:
        return False

    if not getattr(user, "fairdata_username", None):
        return False  # Non-fairdata user, no need to sync

    # No need to sync active user if synced recently
    now = timezone.now()
    if user.is_active and user.synced and (now - user.synced < timezone.timedelta(hours=8)):
        return False

    payload = {
        "id": user.username,
        "token": self.trusted_service_token,
    }
    url = f"{self.host}/user_status"
    res = requests.post(url, payload)
    if res.status_code != 200:
        _logger.warning("User sync error. Failed to get user data from {url}: {res.text} ")
        return False

    # Update user
    data = res.json()
    user.synced = now
    user.is_active = not data["locked"]
    user.csc_projects = data["projects"]
    user.save()
    return True