feat: use token to activate maintenance for rollout

.conf/.env.example

@@ -95,4 +95,5 @@ INFRA_EXPORT_TOKEN=
 INFRA_HADOOP_TOKEN=
 SJS_TOKEN=
 ETL_TOKEN=
+ROLLOUT_MAINTENANCE_TOKEN=
 JWT_SIGNING_KEY=

.conf/.test.env

@@ -90,6 +90,7 @@ INFRA_EXPORT_TOKEN=
 INFRA_HADOOP_TOKEN=
 SJS_TOKEN=RaNdOMkEyToKeNForSJS
 ETL_TOKEN=RaNdOMkEyToKeNForETL
+ROLLOUT_MAINTENANCE_TOKEN=
 JWT_SIGNING_KEY=
 
 INFLUXDB_DISABLED=1

admin_cohort/auth/utils.py

@@ -145,6 +145,8 @@ def get_userinfo_from_token(token: str, auth_method: str) -> Union[None, UserInf
     if token == env("SJS_TOKEN"):
         _logger.info("SJS token connexion")
         return UserInfo.sjs()
+    if token == env("ROLLOUT_MAINTENANCE_TOKEN"):
+        return UserInfo.rollout()
 
     if auth_method == JWT_AUTH_MODE:
         try:

admin_cohort/permissions.py

@@ -3,7 +3,7 @@
 from rest_framework.permissions import OR as drf_OR
 
 from admin_cohort.models import User
-from admin_cohort.settings import ETL_USERNAME, ADMINS
+from admin_cohort.settings import ETL_USERNAME, ADMINS, ROLLOUT_USERNAME
 
 
 def user_is_authenticated(user):
@@ -46,8 +46,8 @@ def has_permission(self, request, view):
         if request.method in permissions.SAFE_METHODS:
             return True
         user = request.user
-        return user_is_authenticated(user) and (user_is_admin(user) or
-                                                user.provider_username == ETL_USERNAME)
+        return user_is_authenticated(user) and \
+            (user_is_admin(user) or user.provider_username in (ROLLOUT_USERNAME, ETL_USERNAME))
 
 
 class LogsPermission(permissions.BasePermission):

admin_cohort/settings.py

@@ -254,6 +254,7 @@
 
 SJS_USERNAME = env("SJS_USERNAME", default="SPARK_JOB_SERVER")
 ETL_USERNAME = env("ETL_USERNAME", default="SOLR_ETL")
+ROLLOUT_USERNAME = env("ROLLOUT_USERNAME", default="ROLLOUT_PIPELINE")
 
 # InfluxDB
 INFLUXDB_DISABLED = int(env("INFLUXDB_DISABLED")) == 1

admin_cohort/types.py

@@ -55,6 +55,13 @@ def sjs(cls):
                    lastname="SERVER",
                    email="[email protected]")
 
+    @classmethod
+    def rollout(cls):
+        return cls(username="ROLLOUT_PIPELINE",
+                   firstname="Rollout",
+                   lastname="PIPELINE",
+                   email="[email protected]")
+
 
 class StrEnum(str, Enum):
     def __str__(self):