Coverage for silkaj/wot/membership.py: 100%

1# Copyright 2016-2024 Maël Azimi <m.a@moul.re> 

2# 

3# Silkaj is free software: you can redistribute it and/or modify 

4# it under the terms of the GNU Affero General Public License as published by 

5# the Free Software Foundation, either version 3 of the License, or 

6# (at your option) any later version. 

7# 

8# Silkaj is distributed in the hope that it will be useful, 

9# but WITHOUT ANY WARRANTY; without even the implied warranty of 

10# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

11# GNU Affero General Public License for more details. 

12# 

13# You should have received a copy of the GNU Affero General Public License 

14# along with Silkaj. If not, see <https://www.gnu.org/licenses/>. 

15 

16import logging 

17import sys 

18 

19import pendulum 

20import rich_click as click 

21from duniterpy.api import bma 

22from duniterpy.documents import BlockID, Membership, get_block_id 

23from duniterpy.key import SigningKey 

24 

25from silkaj import auth, tui 

26from silkaj.blockchain import tools as bc_tools 

27from silkaj.constants import DATE, SUCCESS_EXIT_STATUS 

28from silkaj.g1_monetary_license import license_approval 

29from silkaj.network import client_instance, send_document 

30from silkaj.public_key import gen_pubkey_checksum 

31from silkaj.wot import tools as w_tools 

32 

33 

34@click.command( 

35 "membership", 

36 help="Send and sign membership document: \n\ 

37for first emission and for renewal", 

38) 

39@click.pass_context 

40def send_membership(ctx: click.Context) -> None: 

41 dry_run = ctx.obj["DRY_RUN"] 

42 

43 # Authentication 

44 key = auth.auth_method() 

45 

46 # Get the identity information 

47 head_block = bc_tools.get_head_block() 

48 membership_block_id = BlockID(head_block["number"], head_block["hash"]) 

49 identity = (w_tools.choose_identity(key.pubkey))[0] 

50 identity_uid = identity["uid"] 

51 identity_block_id = get_block_id(identity["meta"]["timestamp"]) 

52 

53 # Display license and ask for confirmation 

54 currency = head_block["currency"] 

55 if not dry_run: 

56 license_approval(currency) 

57 

58 # Confirmation 

59 display_confirmation_table(identity_uid, key.pubkey, identity_block_id) 

60 if not dry_run and not ctx.obj["DISPLAY_DOCUMENT"]: 

61 tui.send_doc_confirmation("membership document for this identity") 

62 

63 # Create and sign membership document 

64 membership = generate_membership_document( 

65 key.pubkey, 

66 membership_block_id, 

67 identity_uid, 

68 identity_block_id, 

69 currency, 

70 key, 

71 ) 

72 

73 logging.debug(membership.signed_raw()) 

74 

75 if dry_run: 

76 click.echo(membership.signed_raw()) 

77 sys.exit(SUCCESS_EXIT_STATUS) 

78 

79 if ctx.obj["DISPLAY_DOCUMENT"]: 

80 click.echo(membership.signed_raw()) 

81 tui.send_doc_confirmation("membership document for this identity") 

82 

83 # Send the membership signed raw document to the node 

84 send_document(bma.blockchain.membership, membership) 

85 

86 

87def display_confirmation_table( 

88 identity_uid: str, 

89 pubkey: str, 

90 identity_block_id: BlockID, 

91) -> None: 

92 """ 

93 Check whether there is pending memberships already in the mempool 

94 Display their expiration date 

95 

96 Actually, sending a membership document works even if the time 

97 between two renewals is not awaited as for the certification 

98 """ 

99 

100 client = client_instance() 

101 

102 identities_requirements = client(bma.wot.requirements, pubkey, pubkey=True) 

103 for identity_requirements in identities_requirements["identities"]: 

104 if identity_requirements["uid"] == identity_uid: 

105 membership_expires = identity_requirements["membershipExpiresIn"] 

106 pending_expires = identity_requirements["membershipPendingExpiresIn"] 

107 pending_memberships = identity_requirements["pendingMemberships"] 

108 break 

109 

110 table = [] 

111 if membership_expires: 

112 expires = pendulum.now().add(seconds=membership_expires).diff_for_humans() 

113 table.append(["Expiration date of current membership", expires]) 

114 

115 if pending_memberships: 

116 table.append( 

117 [ 

118 "Number of pending membership(s) in the mempool", 

119 len(pending_memberships), 

120 ], 

121 ) 

122 

123 table.append( 

124 [ 

125 "Pending membership documents will expire", 

126 pendulum.now().add(seconds=pending_expires).diff_for_humans(), 

127 ], 

128 ) 

129 

130 table.append(["User Identifier (UID)", identity_uid]) 

131 table.append(["Public Key", gen_pubkey_checksum(pubkey)]) 

132 

133 table.append(["Block Identity", str(identity_block_id)[:45] + "…"]) 

134 

135 block = client(bma.blockchain.block, identity_block_id.number) 

136 table.append( 

137 [ 

138 "Identity published", 

139 pendulum.from_timestamp(block["time"], tz="local").format(DATE), 

140 ], 

141 ) 

142 

143 params = bc_tools.get_blockchain_parameters() 

144 table.append( 

145 [ 

146 "Expiration date of new membership", 

147 pendulum.now().add(seconds=params["msValidity"]).diff_for_humans(), 

148 ], 

149 ) 

150 

151 table.append( 

152 [ 

153 "Expiration date of new membership from the mempool", 

154 pendulum.now().add(seconds=params["msPeriod"]).diff_for_humans(), 

155 ], 

156 ) 

157 

158 display_table = tui.Table() 

159 display_table.fill_rows(table) 

160 click.echo(display_table.draw()) 

161 

162 

163def generate_membership_document( 

164 pubkey: str, 

165 membership_block_id: BlockID, 

166 identity_uid: str, 

167 identity_block_id: BlockID, 

168 currency: str, 

169 key: SigningKey = None, 

170) -> Membership: 

171 return Membership( 

172 issuer=pubkey, 

173 membership_block_id=membership_block_id, 

174 uid=identity_uid, 

175 identity_block_id=identity_block_id, 

176 currency=currency, 

177 signing_key=key, 

178 )