/
node_group.py
170 lines (142 loc) · 5.85 KB
/
node_group.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
"""Group of methods for node operation."""
import logging
import pathlib as pl
import typing as tp
from cardano_clusterlib import clusterlib_helpers
from cardano_clusterlib import helpers
from cardano_clusterlib import structs
from cardano_clusterlib import types as itp
LOGGER = logging.getLogger(__name__)
class NodeGroup:
def __init__(self, clusterlib_obj: "itp.ClusterLib") -> None:
self._clusterlib_obj = clusterlib_obj
def gen_kes_key_pair(
self, node_name: str, destination_dir: itp.FileType = "."
) -> structs.KeyPair:
"""Generate a key pair for a node KES operational key.
Args:
node_name: A name of the node the key pair is generated for.
destination_dir: A path to directory for storing artifacts (optional).
Returns:
structs.KeyPair: A tuple containing the key pair.
"""
destination_dir = pl.Path(destination_dir).expanduser()
vkey = destination_dir / f"{node_name}_kes.vkey"
skey = destination_dir / f"{node_name}_kes.skey"
clusterlib_helpers._check_files_exist(vkey, skey, clusterlib_obj=self._clusterlib_obj)
self._clusterlib_obj.cli(
[
"node",
"key-gen-KES",
"--verification-key-file",
str(vkey),
"--signing-key-file",
str(skey),
]
)
helpers._check_outfiles(vkey, skey)
return structs.KeyPair(vkey, skey)
def gen_vrf_key_pair(
self, node_name: str, destination_dir: itp.FileType = "."
) -> structs.KeyPair:
"""Generate a key pair for a node VRF operational key.
Args:
node_name: A name of the node the key pair is generated for.
destination_dir: A path to directory for storing artifacts (optional).
Returns:
structs.KeyPair: A tuple containing the key pair.
"""
destination_dir = pl.Path(destination_dir).expanduser()
vkey = destination_dir / f"{node_name}_vrf.vkey"
skey = destination_dir / f"{node_name}_vrf.skey"
clusterlib_helpers._check_files_exist(vkey, skey, clusterlib_obj=self._clusterlib_obj)
self._clusterlib_obj.cli(
[
"node",
"key-gen-VRF",
"--verification-key-file",
str(vkey),
"--signing-key-file",
str(skey),
]
)
helpers._check_outfiles(vkey, skey)
return structs.KeyPair(vkey, skey)
def gen_cold_key_pair_and_counter(
self, node_name: str, destination_dir: itp.FileType = "."
) -> structs.ColdKeyPair:
"""Generate a key pair for operator's offline key and a new certificate issue counter.
Args:
node_name: A name of the node the key pair and the counter is generated for.
destination_dir: A path to directory for storing artifacts (optional).
Returns:
structs.ColdKeyPair: A tuple containing the key pair and the counter.
"""
destination_dir = pl.Path(destination_dir).expanduser()
vkey = destination_dir / f"{node_name}_cold.vkey"
skey = destination_dir / f"{node_name}_cold.skey"
counter = destination_dir / f"{node_name}_cold.counter"
clusterlib_helpers._check_files_exist(
vkey, skey, counter, clusterlib_obj=self._clusterlib_obj
)
self._clusterlib_obj.cli(
[
"node",
"key-gen",
"--cold-verification-key-file",
str(vkey),
"--cold-signing-key-file",
str(skey),
"--operational-certificate-issue-counter-file",
str(counter),
]
)
helpers._check_outfiles(vkey, skey, counter)
return structs.ColdKeyPair(vkey, skey, counter)
def gen_node_operational_cert(
self,
node_name: str,
kes_vkey_file: itp.FileType,
cold_skey_file: itp.FileType,
cold_counter_file: itp.FileType,
kes_period: tp.Optional[int] = None,
destination_dir: itp.FileType = ".",
) -> pl.Path:
"""Generate a node operational certificate.
This certificate is used when starting the node and not submitted through a transaction.
Args:
node_name: A name of the node the certificate is generated for.
kes_vkey_file: A path to pool KES vkey file.
cold_skey_file: A path to pool cold skey file.
cold_counter_file: A path to pool cold counter file.
kes_period: A start KES period. The current KES period is used when not specified.
destination_dir: A path to directory for storing artifacts (optional).
Returns:
Path: A path to the generated certificate.
"""
destination_dir = pl.Path(destination_dir).expanduser()
out_file = destination_dir / f"{node_name}.opcert"
clusterlib_helpers._check_files_exist(out_file, clusterlib_obj=self._clusterlib_obj)
kes_period = (
kes_period if kes_period is not None else self._clusterlib_obj.g_query.get_kes_period()
)
self._clusterlib_obj.cli(
[
"node",
"issue-op-cert",
"--kes-verification-key-file",
str(kes_vkey_file),
"--cold-signing-key-file",
str(cold_skey_file),
"--operational-certificate-issue-counter",
str(cold_counter_file),
"--kes-period",
str(kes_period),
"--out-file",
str(out_file),
]
)
helpers._check_outfiles(out_file)
return out_file
def __repr__(self) -> str:
return f"<{self.__class__.__name__}: clusterlib_obj={id(self._clusterlib_obj)}>"