Skip to content

feat: Add proxy list, proxy reject, and proxy remove --all #833

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
MkDev11 wants to merge 2 commits into
base: staging
Choose a base branch
from
Open

feat: Add proxy list, proxy reject, and proxy remove --all #833

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
adf8647
feat(proxy): add list, reject, and remove --all (Closes #742)
MkDev11 Feb 26, 2026
20ea617
ci: retrigger CI
MkDev11 Feb 26, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
230 changes: 213 additions & 17 deletions bittensor_cli/cli.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1243,6 +1243,12 @@ def __init__(self):
"execute",
rich_help_panel=HELP_PANELS["PROXY"]["MGMT"],
)(self.proxy_execute_announced)
self.proxy_app.command("list", rich_help_panel=HELP_PANELS["PROXY"]["MGMT"])(
self.proxy_list
)
self.proxy_app.command("reject", rich_help_panel=HELP_PANELS["PROXY"]["MGMT"])(
self.proxy_reject_announced
)

# Sub command aliases
# Wallet
Expand Down Expand Up @@ -9817,15 +9823,23 @@ def proxy_add(
def proxy_remove(
self,
delegate: Annotated[
str,
Optional[str],
typer.Option(
callback=is_valid_ss58_address_param,
prompt="Enter the SS58 address of the delegate to remove, e.g. 5dxds...",
help="The SS58 address of the delegate to remove",
help="The SS58 address of the delegate to remove. Mutually exclusive with --all.",
),
] = "",
] = None,
remove_all: bool = typer.Option(
False,
"--all",
help="Remove all proxies for the account at once.",
),
network: Optional[list[str]] = Options.network,
proxy_type: ProxyType = Options.proxy_type,
proxy_type: ProxyType = typer.Option(
ProxyType.Any.value,
"--proxy-type",
help="Type of proxy (only used with --delegate, not --all)",
),
delay: int = typer.Option(0, help="Delay, in number of blocks"),
wallet_name: str = Options.wallet_name,
wallet_path: str = Options.wallet_path,
Expand All @@ -9842,24 +9856,21 @@ def proxy_remove(
"""
Unregisters a proxy from an account.

Revokes proxy permissions previously granted to another account. This prevents the delegate account from executing any further transactions on your behalf.


[bold]Example:[/bold]
Revoke proxy permissions from a single proxy account
[green]$[/green] btcli proxy remove --delegate 5GDel... --proxy-type Transfer

Revokes proxy permissions previously granted to another account. Use --delegate for a single proxy or --all to remove every proxy.
"""
# TODO should add a --all flag to call Proxy.remove_proxies ?
if remove_all and delegate:
print_error("Cannot use --all and --delegate together. Choose one.")
raise typer.Exit(1)
if not remove_all and not delegate:
print_error("Either --delegate or --all is required.")
raise typer.Exit(1)
logger.debug(
"args:\n"
f"delegate: {delegate}\n"
f"remove_all: {remove_all}\n"
f"network: {network}\n"
f"proxy_type: {proxy_type}\n"
f"delay: {delay}\n"
f"wait_for_finalization: {wait_for_finalization}\n"
f"wait_for_inclusion: {wait_for_inclusion}\n"
f"era: {period}\n"
)
self.verbosity_handler(quiet, verbose, json_output, prompt)
wallet = self.wallet_ask(
Expand All @@ -9869,9 +9880,24 @@ def proxy_remove(
ask_for=[WO.NAME, WO.PATH],
validate=WV.WALLET,
)
subtensor = self.initialize_chain(network)
if remove_all:
return self._run_command(
proxy_commands.remove_all_proxies(
subtensor=subtensor,
wallet=wallet,
prompt=prompt,
decline=decline,
quiet=quiet,
wait_for_inclusion=wait_for_inclusion,
wait_for_finalization=wait_for_finalization,
period=period,
json_output=json_output,
)
)
return self._run_command(
proxy_commands.remove_proxy(
subtensor=self.initialize_chain(network),
subtensor=subtensor,
wallet=wallet,
delegate=delegate,
proxy_type=proxy_type,
Expand Down Expand Up @@ -10195,6 +10221,176 @@ def proxy_execute_announced(
with ProxyAnnouncements.get_db() as (conn, cursor):
ProxyAnnouncements.mark_as_executed(conn, cursor, got_call_from_db)

def proxy_list(
self,
address: Optional[str] = typer.Option(
None,
"--address",
help="SS58 address to list proxies for. Defaults to the selected wallet.",
callback=is_valid_ss58_address_param,
),
network: Optional[list[str]] = Options.network,
wallet_name: Optional[str] = Options.wallet_name,
wallet_path: Optional[str] = Options.wallet_path,
wallet_hotkey: Optional[str] = Options.wallet_hotkey,
quiet: bool = Options.quiet,
verbose: bool = Options.verbose,
json_output: bool = Options.json_output,
):
"""List proxies configured for an account (chain Proxy.Proxies storage)."""
self.verbosity_handler(quiet, verbose, json_output, prompt=False)
if address is None:
wallet = self.wallet_ask(
wallet_name,
wallet_path,
wallet_hotkey,
ask_for=[WO.NAME, WO.PATH],
validate=WV.WALLET,
)
address = wallet.coldkeypub.ss58_address
return self._run_command(
proxy_commands.list_proxies(
subtensor=self.initialize_chain(network),
address=address,
prompt=False,
json_output=json_output,
)
)

def proxy_reject_announced(
self,
delegate: Optional[str] = typer.Option(
None,
"--delegate",
help="SS58 address of the delegate who made the announcement. Required with --no-prompt.",
callback=is_valid_ss58_address_param,
),
call_hash: Optional[str] = typer.Option(
None,
"--call-hash",
help="Hash of the announced call to reject. Resolved from ProxyAnnouncements table if available.",
),
network: Optional[list[str]] = Options.network,
wallet_name: str = Options.wallet_name,
wallet_path: str = Options.wallet_path,
wallet_hotkey: str = Options.wallet_hotkey,
prompt: bool = Options.prompt,
decline: bool = Options.decline,
wait_for_inclusion: bool = Options.wait_for_inclusion,
wait_for_finalization: bool = Options.wait_for_finalization,
period: int = Options.period,
quiet: bool = Options.quiet,
verbose: bool = Options.verbose,
json_output: bool = Options.json_output,
):
"""
Reject a previously announced proxy call (Proxy.reject_announcement).

If --call-hash is provided, the command attempts to resolve it from the
ProxyAnnouncements table and marks it as executed on success (same flow
as btcli proxy execute).
"""
self.verbosity_handler(quiet, verbose, json_output, prompt)
wallet = self.wallet_ask(
wallet_name,
wallet_path,
wallet_hotkey,
ask_for=[WO.NAME, WO.PATH],
validate=WV.WALLET,
)
real_address = wallet.coldkeypub.ss58_address
if delegate is None:
if not prompt:
print_error("--delegate is required when using --no-prompt.")
raise typer.Exit(1)
while True:
delegate = Prompt.ask("Enter the delegate SS58 address")
if is_valid_ss58_address(delegate):
break
print_error(f"Invalid SS58 address: {delegate}")

got_call_from_db: Optional[int] = None
if call_hash is not None:
with ProxyAnnouncements.get_db() as (conn, cursor):
announcements = ProxyAnnouncements.read_rows(conn, cursor)
potential_call_matches = []
for row in announcements:
(
id_,
address,
epoch_time,
block_,
call_hash_,
call_hex_,
call_serialized,
executed_int,
) = row
if (
(call_hash_ == call_hash or f"0x{call_hash_}" == call_hash)
and not bool(executed_int)
and address == real_address
):
potential_call_matches.append(row)
if len(potential_call_matches) == 1:
got_call_from_db = potential_call_matches[0][0]
elif len(potential_call_matches) > 1:
if prompt:
console.print(
f"The call hash you provided matches {len(potential_call_matches)} possible "
"entries for this wallet. The results will be iterated for you to select your intended call."
)
for row in potential_call_matches:
(
id_,
address,
epoch_time,
block_,
call_hash_,
call_hex_,
call_serialized,
executed_int,
) = row
console.print(
f"Time: {datetime.datetime.fromtimestamp(epoch_time)}\nCall:\n"
)
console.print_json(call_serialized)
if confirm_action(
"Is this the intended call?",
decline=decline,
quiet=quiet,
):
got_call_from_db = id_
break
else:
verbose_console.print(
f"Call hash '{call_hash}' matched {len(potential_call_matches)} unexecuted entries "
"for this wallet. Skipping local mark_as_executed due to ambiguity."
)
else:
if not prompt:
print_error("--call-hash is required when using --no-prompt.")
raise typer.Exit(1)
call_hash = Prompt.ask("Enter the call hash to reject")

success = self._run_command(
proxy_commands.reject_announcement(
subtensor=self.initialize_chain(network),
wallet=wallet,
delegate=delegate,
call_hash=call_hash,
prompt=prompt,
decline=decline,
quiet=quiet,
wait_for_inclusion=wait_for_inclusion,
wait_for_finalization=wait_for_finalization,
period=period,
json_output=json_output,
)
)
if success and got_call_from_db is not None:
with ProxyAnnouncements.get_db() as (conn, cursor):
ProxyAnnouncements.mark_as_executed(conn, cursor, got_call_from_db)

@staticmethod
def convert(
from_rao: Optional[str] = typer.Option(
Expand Down
19 changes: 19 additions & 0 deletions bittensor_cli/src/bittensor/extrinsics/mev_shield.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,25 @@ async def encrypt_extrinsic(
return encrypted_call


async def extract_mev_shield_id(response: "AsyncExtrinsicReceipt") -> Optional[str]:
"""
Extract the MEV Shield wrapper ID from an extrinsic response.

After submitting a MEV Shield encrypted call, the EncryptedSubmitted event
contains the wrapper ID needed to track execution.

Args:
response: The extrinsic receipt from submit_extrinsic.

Returns:
The wrapper ID (hex string) or None if not found.
"""
for event in await response.triggered_events:
if event["event_id"] == "EncryptedSubmitted":
return event["attributes"]["id"]
return None


async def wait_for_extrinsic_by_hash(
subtensor: "SubtensorInterface",
extrinsic_hash: str,
Expand Down
Loading
Loading