Coverage for gws-app/gws/base/database/auth_provider.py: 99%

1"""Base authorization provider that checks users with SQL queries.""" 

2 

3from typing import Optional, cast 

4 

5import re 

6 

7import gws 

8import gws.base.auth 

9import gws.base.database.provider 

10import gws.config.util 

11import gws.lib.sa as sa 

12 

13 

14class Config(gws.base.auth.provider.Config): 

15 """SQL-based authorization provider""" 

16 

17 dbUid: Optional[str] 

18 """UID of the database provider.""" 

19 

20 authorizationSql: str 

21 """SQL query that checks user credentials.""" 

22 

23 getUserSql: str 

24 """SQL query that returns the user record for the {uid} placeholder.""" 

25 

26 

27class Placeholders(gws.Enum): 

28 """Placeholder names available in the SQL queries.""" 

29 

30 username = 'username' 

31 """User name from the credentials.""" 

32 password = 'password' 

33 """Password from the credentials.""" 

34 token = 'token' 

35 """Token from the credentials.""" 

36 uid = 'uid' 

37 """Local user id, used in ``getUserSql``.""" 

38 

39 

40class Object(gws.base.auth.provider.Object): 

41 """Authorization provider that checks credentials and loads users with SQL queries.""" 

42 

43 db: gws.DatabaseProvider 

44 """Database provider the queries run on.""" 

45 authorizationSql: str 

46 """SQL query that checks user credentials.""" 

47 getUserSql: str 

48 """SQL query that returns the record of a user by uid.""" 

49 

50 def configure(self): 

51 self.configure_provider() 

52 self.authorizationSql = self.cfg('authorizationSql') 

53 self.getUserSql = self.cfg('getUserSql') 

54 

55 def configure_provider(self): 

56 """Set the database provider from the configuration. 

57 

58 Returns: 

59 ``True`` if a provider was found. 

60 

61 Raises: 

62 ``gws.Error``: If no matching database provider is configured. 

63 """ 

64 return gws.config.util.configure_database_provider_for(self) 

65 

66 def authenticate(self, method, credentials): 

67 params = { 

68 Placeholders.username: credentials.get('username'), 

69 Placeholders.password: credentials.get('password'), 

70 Placeholders.token: credentials.get('token'), 

71 } 

72 

73 rs = self._get_records(self.authorizationSql, params) 

74 

75 if not rs: 

76 return 

77 if len(rs) > 1: 

78 raise gws.ForbiddenError(f'multiple records found') 

79 

80 return self._make_user(rs[0], validate=True) 

81 

82 def get_user(self, local_uid): 

83 params = { 

84 'uid': local_uid, 

85 } 

86 

87 rs = self._get_records(self.getUserSql, params) 

88 

89 if not rs: 

90 return 

91 if len(rs) > 1: 

92 return 

93 

94 return self._make_user(rs[0], validate=False) 

95 

96 def _get_records(self, sql: str, params: dict) -> list[dict]: 

97 """Run a query with ``{name}`` placeholders converted to bind parameters.""" 

98 sql = re.sub(r'{(\w+)}', r':\1', sql) 

99 with self.db.connect() as conn: 

100 return [gws.u.to_dict(r) for r in conn.execute(sa.text(sql), params)] 

101 

102 def _make_user(self, rec: dict, validate: bool) -> gws.User: 

103 """Create a user from a record, checking ``validuser`` and ``validpassword`` if ``validate`` is set.""" 

104 user_rec = {} 

105 

106 valid_user = False 

107 valid_password = False 

108 

109 for k, v in rec.items(): 

110 lk = k.lower() 

111 

112 if lk == 'validuser': 

113 valid_user = bool(v) 

114 elif lk == 'validpassword': 

115 valid_password = bool(v) 

116 elif lk == 'uid': 

117 user_rec['localUid'] = str(v) 

118 else: 

119 user_rec[k] = v 

120 

121 if 'localUid' not in user_rec: 

122 raise gws.ForbiddenError('no uid returned') 

123 

124 if validate and not valid_user: 

125 raise gws.ForbiddenError(f'invalid user') 

126 

127 if validate and not valid_password: 

128 raise gws.ForbiddenError(f'invalid password') 

129 

130 return gws.base.auth.user.from_record(self, user_rec)