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
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
|
#! /usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0-or-later
# depthcharge-tools depthchargectl bless subcommand
# Copyright (C) 2020-2022 Alper Nebi Yasak <alpernebiyasak@gmail.com>
# See COPYRIGHT and LICENSE files for full copyright information.
import argparse
import logging
import subprocess
from depthcharge_tools import __version__
from depthcharge_tools.utils.argparse import (
Command,
Argument,
Group,
CommandExit,
)
from depthcharge_tools.utils.os import (
Disk,
Partition,
CrosPartition,
)
from depthcharge_tools.utils.platform import (
is_cros_boot,
)
from depthcharge_tools.depthchargectl import depthchargectl
@depthchargectl.subcommand("bless")
class depthchargectl_bless(
depthchargectl,
prog="depthchargectl bless",
usage="%(prog)s [options] [DISK | PARTITION]",
add_help=False,
):
"""Set the active or given partition as successfully booted."""
_logger = depthchargectl._logger.getChild("bless")
config_section = "depthchargectl/bless"
@depthchargectl.board.copy()
def board(self, codename=""):
"""Assume we're running on the specified board"""
# We can bless partitions without knowing the board.
try:
return super().board
except Exception as err:
self.logger.warning(err)
return None
@Group
def positionals(self):
"""Positional arguments"""
if self.disk is not None and self.partition is not None:
raise ValueError(
"Disk and partition arguments are mutually exclusive."
)
device = self.disk or self.partition
if isinstance(device, str):
sys_device = self.diskinfo.evaluate(device)
if sys_device is not None:
self.logger.info(
"Using argument '{}' as a block device."
.format(device)
)
device = sys_device
else:
self.logger.info(
"Using argument '{}' as a disk image."
.format(device)
)
device = Disk(device)
if isinstance(device, Disk):
if self.partno is None:
raise ValueError(
"Partno argument is required for disks."
)
partition = device.partition(self.partno)
elif isinstance(device, Partition):
if self.partno is not None and self.partno != device.partno:
raise ValueError(
"Partition and partno arguments are mutually exclusive."
)
partition = device
elif device is None:
self.logger.info(
"No partition given, defaulting to currently booted one."
)
partition = self.diskinfo.by_kern_guid()
if partition is None:
if is_cros_boot():
raise ValueError(
"Couldn't figure out the currently booted partition."
)
else:
raise ValueError(
"A disk or partition argument is required when not "
"booted with depthcharge."
)
self.logger.info(
"Working on partition '{}'."
.format(partition)
)
try:
cros_partitions = partition.disk.cros_partitions()
except subprocess.CalledProcessError as err:
self.logger.debug(
err,
exc_info=self.logger.isEnabledFor(logging.DEBUG),
)
raise ValueError(
"Couldn't get partitions for disk '{}'."
.format(partition.disk)
) from err
if partition not in cros_partitions:
raise ValueError(
"Partition '{}' is not a ChromeOS Kernel partition"
.format(partition)
)
partition = CrosPartition(partition)
self.partition = partition
self.disk = partition.disk
self.partno = partition.partno
@positionals.add
@Argument(nargs=0)
def disk(self, disk=None):
"""Disk image to manage partitions of"""
return disk
@positionals.add
@Argument
def partition(self, partition=None):
"""ChromeOS Kernel partition device to manage"""
return partition
@Group
def options(self):
"""Options"""
@options.add
@Argument("-i", "--partno", nargs=1)
def partno(self, number=None):
"""Partition number in the given disk image"""
try:
if number is not None:
number = int(number)
except:
raise TypeError(
"Partition number must be a positive integer."
)
if number is not None and not number > 0:
raise ValueError(
"Partition number must be a positive integer."
)
return number
@options.add
@Argument("--bad", bad=True)
def bad(self, bad=False):
"""Set the partition as unbootable"""
return bad
@options.add
@Argument("--oneshot", oneshot=True)
def oneshot(self, oneshot=False):
"""Set the partition to be tried once"""
return oneshot
def __call__(self):
if self.bad == False:
try:
self.partition.tries = 1
except subprocess.CalledProcessError as err:
raise CommandExit(
"Failed to set remaining tries for partition '{}'."
.format(self.partition)
) from err
if self.oneshot == False:
try:
self.partition.successful = 1
except subprocess.CalledProcessError as err:
raise CommandExit(
"Failed to set success flag for partition '{}'."
.format(self.partition)
) from err
self.logger.warning(
"Set partition '{}' as successfully booted."
.format(self.partition)
)
else:
try:
self.partition.successful = 0
except subprocess.CalledProcessError as err:
raise CommandExit(
"Failed to unset successful flag for partition '{}'."
.format(self.partition)
) from err
self.logger.warning(
"Set partition '{}' as not yet successfully booted."
.format(self.partition)
)
try:
self.partition.prioritize()
except subprocess.CalledProcessError as err:
raise CommandExit(
"Failed to prioritize partition '{}'."
.format(self.partition)
) from err
self.logger.info(
"Set partition '{}' as the highest-priority bootable part."
.format(self.partition)
)
else:
try:
self.partition.attribute = 0x000
except subprocess.CalledProcessError as err:
raise CommandExit(
"Failed to zero attributes for partition '{}'."
.format(self.partition)
) from err
self.logger.warning(
"Set partition '{}' as a zero-priority unbootable part."
.format(self.partition)
)
global_options = depthchargectl.global_options
config_options = depthchargectl.config_options
|
