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

1# Copyright 2016-2025 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 

17 

18import pendulum 

19import rich_click as click 

20from duniterpy.api import bma 

21from duniterpy.documents import BlockID, Membership, get_block_id 

22from duniterpy.key import SigningKey 

23 

24from silkaj import auth, tui 

25from silkaj.blockchain import tools as bc_tools 

26from silkaj.constants import DATE 

27from silkaj.g1_monetary_license import license_approval 

28from silkaj.network import client_instance, send_document 

29from silkaj.public_key import gen_pubkey_checksum 

30from silkaj.wot import tools as w_tools 

31 

32 

33@click.command("membership", help="Send or renew membership.") 

34@click.pass_context 

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

36 dry_run = ctx.obj["DRY_RUN"] 

37 

38 # Authentication 

39 key = auth.auth_method() 

40 

41 # Get the identity information 

42 head_block = bc_tools.get_head_block() 

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

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

45 identity_uid = identity["uid"] 

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

47 

48 # Display license and ask for confirmation 

49 currency = head_block["currency"] 

50 if not dry_run: 

51 license_approval(currency) 

52 

53 # Confirmation 

54 display_confirmation_table(identity_uid, key.pubkey, identity_block_id) 

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

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

57 

58 # Create and sign membership document 

59 membership = generate_membership_document( 

60 key.pubkey, 

61 membership_block_id, 

62 identity_uid, 

63 identity_block_id, 

64 currency, 

65 key, 

66 ) 

67 

68 logging.debug(membership.signed_raw()) 

69 

70 if dry_run: 

71 click.echo(membership.signed_raw()) 

72 ctx.exit() 

73 

74 if ctx.obj["DISPLAY_DOCUMENT"]: 

75 click.echo(membership.signed_raw()) 

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

77 

78 # Send the membership signed raw document to the node 

79 send_document(bma.blockchain.membership, membership) 

80 

81 

82def display_confirmation_table( 

83 identity_uid: str, 

84 pubkey: str, 

85 identity_block_id: BlockID, 

86) -> None: 

87 """ 

88 Check whether there is pending memberships already in the mempool 

89 Display their expiration date 

90 

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

92 between two renewals is not awaited as for the certification 

93 """ 

94 

95 client = client_instance() 

96 

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

98 for identity_requirements in identities_requirements["identities"]: 

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

100 membership_expires = identity_requirements["membershipExpiresIn"] 

101 pending_expires = identity_requirements["membershipPendingExpiresIn"] 

102 pending_memberships = identity_requirements["pendingMemberships"] 

103 break 

104 

105 table = [] 

106 if membership_expires: 

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

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

109 

110 if pending_memberships: 

111 table.append( 

112 [ 

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

114 len(pending_memberships), 

115 ], 

116 ) 

117 

118 table.append( 

119 [ 

120 "Pending membership documents will expire", 

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

122 ], 

123 ) 

124 

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

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

127 

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

129 

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

131 table.append( 

132 [ 

133 "Identity published", 

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

135 ], 

136 ) 

137 

138 params = bc_tools.get_blockchain_parameters() 

139 table.append( 

140 [ 

141 "Expiration date of new membership", 

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

143 ], 

144 ) 

145 

146 table.append( 

147 [ 

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

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

150 ], 

151 ) 

152 

153 display_table = tui.Table() 

154 display_table.fill_rows(table) 

155 click.echo(display_table.draw()) 

156 

157 

158def generate_membership_document( 

159 pubkey: str, 

160 membership_block_id: BlockID, 

161 identity_uid: str, 

162 identity_block_id: BlockID, 

163 currency: str, 

164 key: SigningKey = None, 

165) -> Membership: 

166 return Membership( 

167 issuer=pubkey, 

168 membership_block_id=membership_block_id, 

169 uid=identity_uid, 

170 identity_block_id=identity_block_id, 

171 currency=currency, 

172 signing_key=key, 

173 )