Authentication refresh and anonymous session support

sanic_security/authentication.py

@@ -143,18 +143,18 @@ async def register(
     """
     email_lower = validate_email(request.form.get("email").lower())
     if await Account.filter(email=email_lower).exists():
-        raise CredentialsError("An account with this email already exists.", 409)
+        raise CredentialsError("An account with this email may already exist.", 409)
     elif await Account.filter(
         username=validate_username(request.form.get("username"))
     ).exists():
-        raise CredentialsError("An account with this username already exists.", 409)
+        raise CredentialsError("An account with this username may already exist.", 409)
     elif (
         request.form.get("phone")
         and await Account.filter(
             phone=validate_phone(request.form.get("phone"))
         ).exists()
     ):
-        raise CredentialsError("An account with this phone number already exists.", 409)
+        raise CredentialsError("An account with this phone number may already exist.", 409)
     validate_password(request.form.get("password"))
     account = await Account.create(
         email=email_lower,
@@ -267,7 +267,8 @@ async def authenticate(request: Request) -> AuthenticationSession:
     """
     authentication_session = await AuthenticationSession.decode(request)
     authentication_session.validate()
-    authentication_session.bearer.validate()
+    if not authentication_session.is_anonymous:
+        authentication_session.bearer.validate()
     return authentication_session
 
 

sanic_security/authorization.py

@@ -57,6 +57,8 @@ async def check_permissions(
         AuthorizationError
     """
     authentication_session = await authenticate(request)
+    if authentication_session.is_anonymous:
+        raise AuthorizationError("Session is anonymous.")
     roles = await authentication_session.bearer.roles.filter(deleted=False).all()
     for role in roles:
         for required_permission, role_permission in zip(
@@ -90,6 +92,8 @@ async def check_roles(request: Request, *required_roles: str) -> AuthenticationS
         AuthorizationError
     """
     authentication_session = await authenticate(request)
+    if authentication_session.is_anonymous:
+        raise AuthorizationError("Session is anonymous.")
     roles = await authentication_session.bearer.roles.filter(deleted=False).all()
     for role in roles:
         if role.name in required_roles:

sanic_security/configuration.py

@@ -38,8 +38,9 @@
     "MAX_CHALLENGE_ATTEMPTS": 5,
     "CAPTCHA_SESSION_EXPIRATION": 60,
     "CAPTCHA_FONT": "captcha-font.ttf",
-    "TWO_STEP_SESSION_EXPIRATION": 200,
-    "AUTHENTICATION_SESSION_EXPIRATION": 2592000,
+    "TWO_STEP_SESSION_EXPIRATION": 300,
+    "AUTHENTICATION_SESSION_EXPIRATION": 86400,
+    "AUTHENTICATION_SESSION_REFRESH": 2592000,
     "ALLOW_LOGIN_WITH_USERNAME": False,
     "INITIAL_ADMIN_EMAIL": "[email protected]",
     "INITIAL_ADMIN_PASSWORD": "admin123",
@@ -63,8 +64,9 @@ class Config(dict):
         MAX_CHALLENGE_ATTEMPTS (str): The maximum amount of session challenge attempts allowed.
         CAPTCHA_SESSION_EXPIRATION (int): The amount of seconds till captcha session expiration on creation. Setting to 0 will disable expiration.
         CAPTCHA_FONT (str): The file path to the font being used for captcha generation.
-        TWO_STEP_SESSION_EXPIRATION (int):  The amount of seconds till two step session expiration on creation. Setting to 0 will disable expiration.
+        TWO_STEP_SESSION_EXPIRATION (int):  The amount of seconds till two-step session expiration on creation. Setting to 0 will disable expiration.
         AUTHENTICATION_SESSION_EXPIRATION (bool): The amount of seconds till authentication session expiration on creation. Setting to 0 will disable expiration.
+        AUTHENTICATION_SESSION_EXPIRATION (bool): The amount of seconds till authentication session refresh expiration.
         ALLOW_LOGIN_WITH_USERNAME (bool): Allows login via username and email.
         INITIAL_ADMIN_EMAIL (str): Email used when creating the initial admin account.
         INITIAL_ADMIN_PASSWORD (str): Password used when creating the initial admin account.
@@ -84,6 +86,7 @@ class Config(dict):
     CAPTCHA_FONT: str
     TWO_STEP_SESSION_EXPIRATION: int
     AUTHENTICATION_SESSION_EXPIRATION: int
+    AUTHENTICATION_SESSION_REFRESH: int
     ALLOW_LOGIN_WITH_USERNAME: bool
     INITIAL_ADMIN_EMAIL: str
     INITIAL_ADMIN_PASSWORD: str

sanic_security/exceptions.py

@@ -123,13 +123,31 @@ def __init__(self, message: str = "Session is deactivated.", code: int = 401):
         super().__init__(message, code)
 
 
+class UnfamiliarLocationError(SessionError):
+    """
+    Raised when session is accessed from an unfamiliar ip address
+    """
+
+    def __init__(self):
+        super().__init__("Session accessed from an unfamiliar location.")
+
+
 class ExpiredError(SessionError):
     """
     Raised when session has expired.
     """
 
     def __init__(self):
-        super().__init__("Session has expired")
+        super().__init__("Session has expired.")
+
+
+class NotExpiredError(SessionError):
+    """
+    Raised when session needs to be expired.
+    """
+
+    def __init__(self):
+        super().__init__("Session has not expired yet.", 403)
 
 
 class SecondFactorRequiredError(SessionError):

sanic_security/models.py

@@ -113,19 +113,6 @@ class Account(BaseModel):
         "models.Role", through="account_role"
     )
 
-    @property
-    def json(self) -> dict:
-        return {
-            "id": self.id,
-            "date_created": str(self.date_created),
-            "date_updated": str(self.date_updated),
-            "email": self.email,
-            "username": self.username,
-            "phone": self.phone,
-            "disabled": self.disabled,
-            "verified": self.verified,
-        }
-
     def validate(self) -> None:
         """
         Raises an error with respect to account state.
@@ -155,6 +142,19 @@ async def disable(self):
             self.disabled = True
             await self.save(update_fields=["disabled"])
 
+    @property
+    def json(self) -> dict:
+        return {
+            "id": self.id,
+            "date_created": str(self.date_created),
+            "date_updated": str(self.date_updated),
+            "email": self.email,
+            "username": self.username,
+            "phone": self.phone,
+            "disabled": self.disabled,
+            "verified": self.verified,
+        }
+
     @staticmethod
     async def get_via_email(email: str):
         """
@@ -237,19 +237,6 @@ class Session(BaseModel):
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
 
-    @property
-    def json(self) -> dict:
-        return {
-            "id": self.id,
-            "date_created": str(self.date_created),
-            "date_updated": str(self.date_updated),
-            "expiration_date": str(self.expiration_date),
-            "bearer": self.bearer.username
-            if isinstance(self.bearer, Account)
-            else None,
-            "active": self.active,
-        }
-
     def validate(self) -> None:
         """
         Raises an error with respect to session state.
@@ -261,13 +248,10 @@ def validate(self) -> None:
         """
         if self.deleted:
             raise DeletedError("Session has been deleted.")
-        elif (
-            self.expiration_date
-            and datetime.datetime.now(datetime.timezone.utc) >= self.expiration_date
-        ):
-            raise ExpiredError()
         elif not self.active:
             raise DeactivatedError()
+        elif self.expiration_date and datetime.datetime.now(datetime.timezone.utc) >= self.expiration_date:
+            raise ExpiredError()
 
     async def deactivate(self):
         """
@@ -314,12 +298,49 @@ def encode(self, response: HTTPResponse) -> None:
         if security_config.SESSION_DOMAIN:
             response.cookies.get_cookie(cookie).domain = security_config.SESSION_DOMAIN
 
+    async def location_familiarity(self, request):
+        """
+        Checks if the client is accessing session from a familiar location.
+
+        Args:
+            request (Request): Sanic request parameter.
+
+        Raises:
+            UnfamiliarLocationError
+        """
+        client_ip = get_ip(request)
+        if client_ip != self.ip and not await self.filter(ip=client_ip, deleted=False).exists():
+            raise UnfamiliarLocationError()
+
+    @property
+    def is_anonymous(self) -> bool:
+        """
+        Determines if an account is associated with session.
+
+        Returns:
+            is_anonymous
+        """
+        return self.bearer is None
+
+    @property
+    def json(self) -> dict:
+        return {
+            "id": self.id,
+            "date_created": str(self.date_created),
+            "date_updated": str(self.date_updated),
+            "expiration_date": str(self.expiration_date),
+            "bearer": self.bearer.username
+            if isinstance(self.bearer, Account)
+            else None,
+            "active": self.active,
+        }
+
     @classmethod
     async def new(
-        cls,
-        request: Request,
-        account: Account,
-        **kwargs: Union[int, str, bool, float, list, dict],
+            cls,
+            request: Request,
+            account: Account,
+            **kwargs: Union[int, str, bool, float, list, dict],
     ):
         """
         Creates session with pre-set values.
@@ -348,7 +369,7 @@ async def get_associated(cls, account: Account):
         Raises:
             NotFoundError
         """
-        sessions = await cls.filter(bearer=account).prefetch_related("bearer").all()
+        sessions = await cls.filter(bearer=account, deleted=False).all()
         if not sessions:
             raise NotFoundError("No sessions associated to account were found.")
         return sessions
@@ -375,7 +396,8 @@ def decode_raw(cls, request: Request) -> dict:
                 raise JWTDecodeError("Session token not provided or expired.", 401)
             else:
                 return jwt.decode(
-                    cookie, security_config.PUBLIC_SECRET or security_config.SECRET, algorithms=[security_config.SESSION_ENCODING_ALGORITHM]
+                    cookie, security_config.PUBLIC_SECRET or security_config.SECRET,
+                    security_config.SESSION_ENCODING_ALGORITHM
                 )
         except DecodeError as e:
             raise JWTDecodeError(str(e))
@@ -516,6 +538,7 @@ class AuthenticationSession(Session):
     """
 
     requires_second_factor: bool = fields.BooleanField(default=False)
+    refresh_date: datetime.datetime = fields.DatetimeField(null=True)
 
     def validate(self) -> None:
         """
@@ -531,6 +554,28 @@ def validate(self) -> None:
         if self.requires_second_factor:
             raise SecondFactorRequiredError()
 
+    async def refresh(self, request: Request):
+        """
+        Seamlessly creates new session if within refresh date.
+
+        Raises:
+            DeletedError
+            ExpiredError
+            DeactivatedError
+            SecondFactorRequiredError
+            NotExpiredError
+        """
+        try:
+            self.validate()
+            raise NotExpiredError()
+        except ExpiredError as e:
+            if datetime.datetime.now(datetime.timezone.utc) <= self.refresh_date:
+                self.active = False
+                await self.save(update_fields=["active"])
+                return self.new(request, self.bearer)
+            else:
+                raise e
+
     @classmethod
     async def new(cls, request: Request, account: Account, **kwargs):
         return await AuthenticationSession.create(
@@ -540,6 +585,9 @@ async def new(cls, request: Request, account: Account, **kwargs):
             expiration_date=get_expiration_date(
                 security_config.AUTHENTICATION_SESSION_EXPIRATION
             ),
+            refresh_date=get_expiration_date(
+                security_config.AUTHENTICATION_SESSION_REFRESH
+            )
         )
 
     class Meta:

sanic_security/test/server.py

@@ -248,16 +248,12 @@ async def on_account_creation(request):
     """
     Quick account creation.
     """
-    if await Account.filter(email=request.form.get("email").lower()).exists():
-        raise CredentialsError("An account with this email already exists.", 409)
-    elif await Account.filter(username=request.form.get("username")).exists():
-        raise CredentialsError("An account with this username already exists.", 409)
     account = await Account.create(
         username=request.form.get("username"),
         email=request.form.get("email").lower(),
         password=password_hasher.hash("password"),
         verified=True,
-        dbisabled=False,
+        disabled=False,
     )
     response = json("Account creation successful!", account.json)
     return response