Coverage for gws-app/gws/plugin/qfieldcloud/auth.py: 98%

1"""Token authorization method for QField clients.""" 

2 

3import datetime 

4import re 

5 

6import gws 

7import gws.base.auth 

8import gws.lib.datetimex as dtx 

9 

10 

11@gws.ext.config.authMethod('qfieldcloud') 

12class Config(gws.base.auth.method.Config): 

13 """Token authentication for QField clients.""" 

14 

15 

16class Result(gws.Data): 

17 """Result of a successful authentication.""" 

18 

19 sess: gws.AuthSession 

20 """Session of the user.""" 

21 user: gws.User 

22 """Authenticated user.""" 

23 token: str 

24 """Session token, sent by QField in the ``Authorization`` header.""" 

25 expiresAt: datetime.datetime 

26 """Expiry time of the session.""" 

27 

28 

29@gws.ext.object.authMethod('qfieldcloud') 

30class Object(gws.base.auth.method.Object): 

31 """QField Cloud authorization method. 

32 

33 Token authorization for QField clients. The ``qfieldcloud`` action creates 

34 sessions with this method on login and accepts only tokens of its sessions. 

35 """ 

36 

37 def configure(self): 

38 self.uid = 'gws.plugin.qfieldcloud.auth' 

39 

40 def authenticate_from_credentials(self, credentials: gws.Data, req: gws.WebRequester) -> Result: 

41 """Authenticate a user by username and password and create a session. 

42 

43 Args: 

44 credentials: Data with ``username`` and ``password``. 

45 req: Web request. 

46 

47 Returns: 

48 Authentication result with the new session and its expiry time. 

49 

50 Raises: 

51 ``gws.AuthenticationError``: If the credentials are invalid. 

52 """ 

53 am = self.root.app.authMgr 

54 user = am.authenticate(self, credentials, req) 

55 if not user: 

56 raise gws.AuthenticationError('invalid username or password') 

57 

58 sess = am.sessionMgr.create(self, user) 

59 return Result( 

60 sess=sess, 

61 user=user, 

62 token=sess.uid, 

63 expiresAt=dtx.add(seconds=am.sessionMgr.lifeTime), 

64 ) 

65 

66 def authenticate_from_token(self, req: gws.WebRequester) -> Result: 

67 """Authenticate a request by the ``Authorization: Token ...`` header and touch the session. 

68 

69 Args: 

70 req: Web request. 

71 

72 Returns: 

73 Authentication result with the existing session and its new expiry time. 

74 

75 Raises: 

76 ``gws.AuthenticationError``: If the header is missing, or the token is invalid or belongs to another method. 

77 ``gws.ForbiddenError``: If the method cannot be used in this context, e.g. without a secure connection. 

78 """ 

79 h = req.header('Authorization', '') 

80 m = re.match(r'^Token (.+)$', h) 

81 if not m: 

82 raise gws.AuthenticationError('qfieldcloud_auth: missing or invalid Authorization header') 

83 

84 token = m.group(1) 

85 am = self.root.app.authMgr 

86 if not am.can_use_method(req, self): 

87 raise gws.ForbiddenError('qfieldcloud_auth: insecure_context') 

88 

89 sess = am.sessionMgr.get(token) 

90 if not sess: 

91 raise gws.AuthenticationError('qfieldcloud_auth: invalid or expired token') 

92 if not sess.method or sess.method.uid != self.uid: 

93 raise gws.AuthenticationError(f'qfieldcloud_auth: wrong method {sess.method=}') 

94 

95 am.sessionMgr.touch(sess) 

96 gws.log.debug(f'qfieldcloud_auth: ok: {token=} {sess.user.uid=} {sess.user.loginName=}') 

97 

98 return Result( 

99 sess=sess, 

100 user=sess.user, 

101 token=sess.uid, 

102 expiresAt=dtx.add(seconds=am.sessionMgr.lifeTime), 

103 )