Coverage for silkaj/wot/certification.py: 29%

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 sys 

17 

18import rich_click as click 

19from duniterpy.api import bma 

20from duniterpy.api.client import Client 

21from duniterpy.documents import Block, BlockID, Certification, Identity, get_block_id 

22from duniterpy.key import SigningKey 

23from pendulum import from_timestamp, now 

24 

25from silkaj import tui 

26from silkaj.auth import auth_method 

27from silkaj.blockchain import tools as bc_tools 

28from silkaj.constants import ALL, DATE 

29from silkaj.g1_monetary_license import license_approval 

30from silkaj.network import client_instance, send_document 

31from silkaj.public_key import gen_pubkey_checksum, is_pubkey_and_check 

32from silkaj.wot import tools as wot_tools 

33 

34 

35@click.command("certify", help="Send certification") 

36@click.argument("uid_pubkey_to_certify") 

37@click.pass_context 

38def certify(ctx: click.Context, uid_pubkey_to_certify: str) -> None: 

39 client = client_instance() 

40 

41 checked_pubkey = is_pubkey_and_check(uid_pubkey_to_certify) 

42 if checked_pubkey: 

43 uid_pubkey_to_certify = str(checked_pubkey) 

44 

45 idty_to_certify, pubkey_to_certify, send_certs = wot_tools.choose_identity( 

46 uid_pubkey_to_certify, 

47 ) 

48 

49 # Authentication 

50 key = auth_method() 

51 

52 issuer_pubkey = key.pubkey 

53 issuer = pre_checks(client, issuer_pubkey, pubkey_to_certify) 

54 

55 # Display license and ask for confirmation 

56 head = bc_tools.get_head_block() 

57 currency = head["currency"] 

58 license_approval(currency) 

59 

60 # Certification confirmation 

61 certification_confirmation( 

62 ctx, 

63 issuer, 

64 issuer_pubkey, 

65 pubkey_to_certify, 

66 idty_to_certify, 

67 ) 

68 

69 # Create and sign certification document 

70 certification = docs_generation( 

71 currency, 

72 pubkey_to_certify, 

73 idty_to_certify, 

74 issuer_pubkey, 

75 head, 

76 key, 

77 ) 

78 

79 if ctx.obj["DISPLAY_DOCUMENT"]: 

80 click.echo(certification.signed_raw(), nl=False) 

81 tui.send_doc_confirmation("certification") 

82 

83 # Send certification document 

84 send_document(bma.wot.certify, certification) 

85 

86 

87def pre_checks(client: Client, issuer_pubkey: str, pubkey_to_certify: str) -> dict: 

88 # Check whether current user is member 

89 issuer = wot_tools.is_member(issuer_pubkey) 

90 if not issuer: 

91 sys.exit("Current identity is not member.") 

92 

93 if issuer_pubkey == pubkey_to_certify: 

94 sys.exit("You can`t certify yourself!") 

95 

96 # Check if the certification can be renewed 

97 params = bc_tools.get_blockchain_parameters() 

98 requirements = client(bma.wot.requirements, pubkey_to_certify, pubkey=True) 

99 req = requirements["identities"][0] # type: dict 

100 for cert in req["certifications"]: 

101 if cert["from"] == issuer_pubkey: 

102 # Ğ1: 0<->2y - 2y + 2m 

103 # ĞT: 0<->4.8m - 4.8m + 12.5d 

104 renewable = cert["expiresIn"] - params["sigValidity"] + params["sigReplay"] 

105 if renewable > 0: 

106 renewable_date = now().add(seconds=renewable).format(DATE) 

107 sys.exit(f"Certification renewable from {renewable_date}") 

108 

109 # Check if the certification is already in the pending certifications 

110 for pending_cert in req["pendingCerts"]: 

111 if pending_cert["from"] == issuer_pubkey: 

112 sys.exit("Certification is currently being processed") 

113 return issuer 

114 

115 

116def certification_confirmation( 

117 ctx: click.Context, 

118 issuer: dict, 

119 issuer_pubkey: str, 

120 pubkey_to_certify: str, 

121 idty_to_certify: dict, 

122) -> None: 

123 cert = [] 

124 client = client_instance() 

125 idty_timestamp = idty_to_certify["meta"]["timestamp"] 

126 block_id_idty = get_block_id(idty_timestamp) 

127 block = client(bma.blockchain.block, block_id_idty.number) 

128 timestamp_date = from_timestamp(block["time"], tz="local").format(ALL) 

129 block_id_date = f": #{idty_timestamp[:15]}… {timestamp_date}" 

130 cert.append(["ID", issuer["uid"], "->", idty_to_certify["uid"] + block_id_date]) 

131 cert.append( 

132 [ 

133 "Pubkey", 

134 gen_pubkey_checksum(issuer_pubkey), 

135 "->", 

136 gen_pubkey_checksum(pubkey_to_certify), 

137 ], 

138 ) 

139 params = bc_tools.get_blockchain_parameters() 

140 cert_ends = now().add(seconds=params["sigValidity"]).format(DATE) 

141 cert.append(["Valid", now().format(DATE), "—>", cert_ends]) 

142 

143 table = tui.Table() 

144 table.fill_rows( 

145 cert, 

146 ["Cert", "Issuer", "->", "Recipient: Published: #block-hash date"], 

147 ) 

148 click.echo(table.draw()) 

149 

150 if not ctx.obj["DISPLAY_DOCUMENT"]: 

151 tui.send_doc_confirmation("certification") 

152 

153 

154def docs_generation( 

155 currency: str, 

156 pubkey_to_certify: str, 

157 idty_to_certify: dict, 

158 issuer_pubkey: str, 

159 head: Block, 

160 key: SigningKey, 

161) -> Certification: 

162 identity = Identity( 

163 pubkey=pubkey_to_certify, 

164 uid=idty_to_certify["uid"], 

165 block_id=get_block_id(idty_to_certify["meta"]["timestamp"]), 

166 currency=currency, 

167 ) 

168 identity.signature = idty_to_certify["self"] 

169 

170 return Certification( 

171 pubkey_from=issuer_pubkey, 

172 identity=identity, 

173 block_id=BlockID(head["number"], head["hash"]), 

174 signing_key=key, 

175 currency=currency, 

176 )