Coverage for gws-app/gws/base/auth/cli.py: 0%

1"""CLI commands for authorization sessions.""" 

2 

3from typing import Optional, cast 

4 

5import gws 

6import gws.lib.datetimex as dtx 

7import gws.lib.cli as cli 

8 

9from . import manager 

10 

11 

12class RemoveParams(gws.CliParams): 

13 """Parameters of the ``authSessrem`` command.""" 

14 

15 older: Optional[int] 

16 """Remove sessions older than N seconds.""" 

17 uid: Optional[list[str]] 

18 """Remove specific sessions.""" 

19 all: Optional[bool] 

20 """Remove all sessions.""" 

21 

22 

23@gws.ext.object.cli('auth') 

24class Object(gws.Node): 

25 """CLI commands for authorization sessions. 

26 

27 Lists the active sessions and removes sessions from the session store. 

28 """ 

29 

30 @gws.ext.command.cli('authSessions') 

31 def sessions(self, p: gws.EmptyRequest): 

32 """Show active authorization sessions.""" 

33 

34 root = gws.load_root() 

35 sm = root.app.authMgr.sessionMgr 

36 

37 sm.cleanup() 

38 

39 rows = [] 

40 

41 for s in sm.list_all(): 

42 dc = dtx.total_difference(s.created).seconds 

43 du = dtx.total_difference(s.updated).seconds 

44 dd = dtx.total_difference(s.created, s.updated).seconds 

45 rows.append((du, dict( 

46 uid=s.uid, 

47 user=s.user.loginName, 

48 started=dtx.to_iso_string(s.created, sep=' ') + f' ({dtx.format_duration(dc)})', 

49 updated=dtx.to_iso_string(s.updated, sep=' ') + f' ({dtx.format_duration(du)})', 

50 duration=dtx.format_duration(dd), 

51 ))) 

52 

53 # oldest first 

54 rows.sort(key=lambda r: -r[0]) 

55 

56 cli.info('') 

57 cli.info(f'Active sessions: {len(rows)}') 

58 cli.info(cli.text_table([r[1] for r in rows], header='auto')) 

59 

60 @gws.ext.command.cli('authSessrem') 

61 def sessrem(self, p: RemoveParams): 

62 """Remove authorization sessions.""" 

63 

64 root = gws.load_root() 

65 sm = root.app.authMgr.sessionMgr 

66 

67 n = 0 

68 

69 for s in sm.list_all(): 

70 du = dtx.total_difference(s.updated).seconds 

71 if p.all: 

72 sm.delete(s) 

73 n += 1 

74 continue 

75 if p.older and du > p.older: 

76 sm.delete(s) 

77 n += 1 

78 continue 

79 if p.uid and s.uid in p.uid: 

80 sm.delete(s) 

81 n += 1 

82 continue 

83 

84 cli.info('') 

85 cli.info(f'Removed sessions: {n}')