weegee: cli: factor out common subcommands
This commit is contained in:
parent
104bbe8cfd
commit
45b3d5b189
|
|
@ -49,9 +49,14 @@ def timestampify(dt: datetime) -> str:
|
|||
|
||||
return f'{round(n)} {unit}{"s" if n >= 1.5 else ""} {"from now" if future else "ago"}'
|
||||
|
||||
def do_get(item: Any, var: str) -> None:
|
||||
def perform_get(item: Any, var: str) -> None:
|
||||
for piece in var.split('.'):
|
||||
item = getattr(item, piece)
|
||||
base, *keys = piece.split('/')
|
||||
if base:
|
||||
item = getattr(item, base)
|
||||
for k in keys:
|
||||
item = item[int(k)]
|
||||
|
||||
if item is None:
|
||||
print()
|
||||
elif isinstance(item, list):
|
||||
|
|
@ -272,10 +277,33 @@ def main():
|
|||
sys_clean.set_defaults(func=do_clean, parser=sys_clean)
|
||||
|
||||
|
||||
# Common subcommands
|
||||
|
||||
def add_common_subcommands(sparser: argparse._SubParsersAction) -> None:
|
||||
def do_get(parser: argparse.ArgumentParser, args: argparse.Namespace, ctx: WeegeeContext) -> None:
|
||||
item = args.objtype.load(ctx, args.name)
|
||||
for var in args.var:
|
||||
perform_get(item, var)
|
||||
|
||||
get = sparser.add_parser('get', help='get variable')
|
||||
get.add_argument('name', help='name')
|
||||
get.add_argument('var', nargs='*', help='variable spec')
|
||||
get.set_defaults(func=do_get, parser=get)
|
||||
|
||||
def do_delete(parser: argparse.ArgumentParser, args: argparse.Namespace, ctx: WeegeeContext) -> None:
|
||||
if not args.yes_i_want_to_destroy_this:
|
||||
parser.error('please pass --yes-i-want-to-destroy-this if you really want to destroy')
|
||||
item = args.objtype.load(ctx, args.name)
|
||||
item.delete()
|
||||
|
||||
delete = sparser.add_parser('destroy', help='delete')
|
||||
delete.add_argument('name', help='host name')
|
||||
delete.set_defaults(func=do_delete, parser=delete)
|
||||
|
||||
# Host commands
|
||||
|
||||
host = commands.add_parser('host', help='host commands')
|
||||
host.set_defaults(parser=host)
|
||||
host.set_defaults(parser=host, objtype=WeegeeHost)
|
||||
host_commands = host.add_subparsers(title='subcommand')
|
||||
|
||||
def do_add_host(parser: argparse.ArgumentParser, args: argparse.Namespace, ctx: WeegeeContext) -> None:
|
||||
|
|
@ -297,15 +325,7 @@ def main():
|
|||
add_host.add_argument('name', help='host name')
|
||||
add_host.set_defaults(func=do_add_host, parser=add_host)
|
||||
|
||||
def do_del_host(parser: argparse.ArgumentParser, args: argparse.Namespace, ctx: WeegeeContext) -> None:
|
||||
if not args.yes_i_want_to_destroy_this:
|
||||
parser.error('please pass --yes-i-want-to-destroy-this if you really want to destroy this host')
|
||||
host = WeegeeHost.load(ctx, args.name)
|
||||
host.delete()
|
||||
|
||||
del_host = host_commands.add_parser('destroy', help='delete host')
|
||||
del_host.add_argument('name', help='host name')
|
||||
del_host.set_defaults(func=do_del_host, parser=del_host)
|
||||
add_common_subcommands(host_commands)
|
||||
|
||||
def do_configure_host(parser: argparse.ArgumentParser, args: argparse.Namespace, ctx: WeegeeContext) -> None:
|
||||
host = WeegeeHost.load(ctx, args.name)
|
||||
|
|
@ -395,7 +415,7 @@ def main():
|
|||
# Interface commands
|
||||
|
||||
interface = commands.add_parser('interface', help='interface commands')
|
||||
interface.set_defaults(parser=interface)
|
||||
interface.set_defaults(parser=interface, objtype=WeegeeInterface)
|
||||
interface_commands = interface.add_subparsers(title='subcommand')
|
||||
|
||||
def do_add_interface(parser: argparse.ArgumentParser, args: argparse.Namespace, ctx: WeegeeContext) -> None:
|
||||
|
|
@ -418,25 +438,7 @@ def main():
|
|||
add_interface.add_argument('name', help='interface name')
|
||||
add_interface.set_defaults(func=do_add_interface, parser=add_interface)
|
||||
|
||||
def do_del_interface(parser: argparse.ArgumentParser, args: argparse.Namespace, ctx: WeegeeContext) -> None:
|
||||
if not args.yes_i_want_to_destroy_this:
|
||||
parser.error('please pass --yes-i-want-to-destroy-this if you really want to destroy this interface')
|
||||
interface = WeegeeInterface.load(ctx, args.name)
|
||||
interface.delete()
|
||||
|
||||
del_interface = interface_commands.add_parser('destroy', help='delete interface')
|
||||
del_interface.add_argument('name', help='interface name')
|
||||
del_interface.set_defaults(func=do_del_interface, parser=del_interface)
|
||||
|
||||
def do_get_interface(parser: argparse.ArgumentParser, args: argparse.Namespace, ctx: WeegeeContext) -> None:
|
||||
interface = WeegeeInterface.load(ctx, args.name)
|
||||
for var in args.var:
|
||||
do_get(interface, var)
|
||||
|
||||
get_interface = interface_commands.add_parser('get', help='get interface variable')
|
||||
get_interface.add_argument('name', help='interface name')
|
||||
get_interface.add_argument('var', nargs='*', help='variable spec')
|
||||
get_interface.set_defaults(func=do_get_interface, parser=get_interface)
|
||||
add_common_subcommands(interface_commands)
|
||||
|
||||
def do_conf_interface(parser: argparse.ArgumentParser, args: argparse.Namespace, ctx: WeegeeContext) -> None:
|
||||
interface = WeegeeInterface.load(ctx, args.name)
|
||||
|
|
@ -484,8 +486,78 @@ def main():
|
|||
|
||||
# Peer commands
|
||||
|
||||
def add_common_peer_subcommands(sparser: argparse._SubParsersAction) -> None:
|
||||
add_common_subcommands(sparser)
|
||||
|
||||
def do_export(parser: argparse.ArgumentParser, args: argparse.Namespace, ctx: WeegeeContext) -> None:
|
||||
peer = args.objtype.load(ctx, args.name)
|
||||
interface = peer.interface.to_public()
|
||||
|
||||
print(f'#+++ {interface.item_name}')
|
||||
for l in interface.item.purify().dump():
|
||||
print(l)
|
||||
print(f'#+++ {peer.item_name}')
|
||||
for l in peer.item.purify().dump():
|
||||
print(l)
|
||||
|
||||
export = sparser.add_parser('export', help='export peer')
|
||||
export.add_argument('name', help='name')
|
||||
export.set_defaults(func=do_export, parser=export)
|
||||
|
||||
def do_connect(parser: argparse.ArgumentParser, args: argparse.Namespace, ctx: WeegeeContext) -> None:
|
||||
if not WeegeeConnection.exists(ctx, args.name):
|
||||
connection = WeegeeConnection.create(ctx, args.name,
|
||||
peers=[], preshared_key=args.preshared_key,
|
||||
)
|
||||
else:
|
||||
connection = WeegeeConnection.load(ctx, args.name)
|
||||
connection.peers += [WeegeePeer.load(ctx, peer).item for peer in args.peer]
|
||||
if args.preshared_key:
|
||||
connection.preshared_key = preshared_key
|
||||
connection.save()
|
||||
|
||||
connect = sparser.add_parser('connect', help='connect peers')
|
||||
connect.add_argument('-k', '--preshared-key', metavar='NAME', help='preshared key (optional)')
|
||||
connect.add_argument('name', help='connection name')
|
||||
connect.add_argument('peer', nargs='*', help='peer names')
|
||||
connect.set_defaults(func=do_connect, parser=connect)
|
||||
|
||||
def do_disconnect(parser: argparse.ArgumentParser, args: argparse.Namespace, ctx: WeegeeContext) -> None:
|
||||
connection = args.objtype.load(ctx, args.name)
|
||||
peers = connection.peers
|
||||
for peer in args.peer:
|
||||
peers.remove(WeegeePeer.load(ctx, peer).item)
|
||||
connection.peers = peers
|
||||
if connection.peers:
|
||||
connection.save()
|
||||
else:
|
||||
connection.delete()
|
||||
|
||||
disconnect = sparser.add_parser('disconnect', help='disconnect peers')
|
||||
disconnect.add_argument('name', help='connection name')
|
||||
disconnect.add_argument('peer', nargs='+', help='peer names')
|
||||
disconnect.set_defaults(func=do_disconnect, parser=disconnect)
|
||||
|
||||
def do_get_connection(parser: argparse.ArgumentParser, args: argparse.Namespace, ctx: WeegeeContext) -> None:
|
||||
connection = args.objtype.load(ctx, args.name)
|
||||
for var in args.var:
|
||||
perform_get(connection, var)
|
||||
|
||||
get_connection = sparser.add_parser('get-connect', help='get connection variable')
|
||||
get_connection.add_argument('name', help='connection name')
|
||||
get_connection.add_argument('var', nargs='*', help='variable spec')
|
||||
get_connection.set_defaults(func=do_get_connection, parser=get_connection)
|
||||
|
||||
def do_print_config(parser: argparse.ArgumentParser, args: argparse.Namespace, ctx: WeegeeContext) -> None:
|
||||
peer = args.objtype.load(ctx, args.name)
|
||||
print(config_interface(peer.interface.to_full()))
|
||||
|
||||
print_config = sparser.add_parser('print-config', help='show WireGuard configuration')
|
||||
print_config.add_argument('name', help='peer name')
|
||||
print_config.set_defaults(func=do_print_config, parser=print_config)
|
||||
|
||||
peer = commands.add_parser('peer', help='peer commands')
|
||||
peer.set_defaults(parser=peer)
|
||||
peer.set_defaults(parser=peer, objtype=WeegeePeer)
|
||||
peer_commands = peer.add_subparsers(title='subcommand')
|
||||
|
||||
def do_add_peer(parser: argparse.ArgumentParser, args: argparse.Namespace, ctx: WeegeeContext) -> None:
|
||||
|
|
@ -505,90 +577,13 @@ def main():
|
|||
add_peer.add_argument('interface', help='existing interface name')
|
||||
add_peer.set_defaults(func=do_add_peer, parser=add_peer)
|
||||
|
||||
def do_del_peer(parser: argparse.ArgumentParser, args: argparse.Namespace, ctx: WeegeeContext) -> None:
|
||||
if not args.yes_i_want_to_destroy_this:
|
||||
parser.error('please pass --yes-i-want-to-destroy-this if you really want to destroy this peer')
|
||||
peer = WeegeePeer.load(ctx, args.name)
|
||||
peer.delete()
|
||||
|
||||
del_peer = peer_commands.add_parser('destroy', help='delete peer')
|
||||
del_peer.add_argument('name', help='peer name')
|
||||
del_peer.set_defaults(func=do_del_peer, parser=del_peer)
|
||||
|
||||
def do_get_peer(parser: argparse.ArgumentParser, args: argparse.Namespace, ctx: WeegeeContext) -> None:
|
||||
peer = WeegeePeer.load(ctx, args.name)
|
||||
for var in args.var:
|
||||
do_get(peer, var)
|
||||
|
||||
get_peer = peer_commands.add_parser('get', help='get peer variable')
|
||||
get_peer.add_argument('name', help='peer name')
|
||||
get_peer.add_argument('var', nargs='*', help='variable spec')
|
||||
get_peer.set_defaults(func=do_get_peer, parser=get_peer)
|
||||
|
||||
def do_export_peer(parser: argparse.ArgumentParser, args: argparse.Namespace, ctx: WeegeeContext) -> None:
|
||||
peer = WeegeePeer.load(ctx, args.name)
|
||||
interface = peer.interface.to_public()
|
||||
|
||||
print(f'#+++ {interface.item_name}')
|
||||
for l in interface.item.purify().dump():
|
||||
print(l)
|
||||
print(f'#+++ {peer.item_name}')
|
||||
for l in peer.item.purify().dump():
|
||||
print(l)
|
||||
|
||||
export_peer = peer_commands.add_parser('export', help='export public part of peer')
|
||||
export_peer.add_argument('name', help='peer name')
|
||||
export_peer.set_defaults(func=do_export_peer, parser=export_peer)
|
||||
|
||||
def do_connect_peer(parser: argparse.ArgumentParser, args: argparse.Namespace, ctx: WeegeeContext) -> None:
|
||||
if not WeegeeConnection.exists(ctx, args.name):
|
||||
connection = WeegeeConnection.create(ctx, args.name,
|
||||
peers=[], preshared_key=args.preshared_key,
|
||||
)
|
||||
else:
|
||||
connection = WeegeeConnection.load(ctx, args.name)
|
||||
connection.peers += [WeegeePeer.load(ctx, peer).item for peer in args.peer]
|
||||
if args.preshared_key:
|
||||
connection.preshared_key = preshared_key
|
||||
connection.save()
|
||||
|
||||
connect_peer = peer_commands.add_parser('connect', help='connect peers')
|
||||
connect_peer.add_argument('-k', '--preshared-key', metavar='NAME', help='preshared key (optional)')
|
||||
connect_peer.add_argument('name', help='connection name')
|
||||
connect_peer.add_argument('peer', nargs='*', help='peer names')
|
||||
connect_peer.set_defaults(func=do_connect_peer, parser=connect_peer)
|
||||
|
||||
def do_disconnect_peer(parser: argparse.ArgumentParser, args: argparse.Namespace, ctx: WeegeeContext) -> None:
|
||||
connection = WeegeeConnection.load(ctx, args.name)
|
||||
peers = connection.peers
|
||||
for peer in args.peer:
|
||||
peers.remove(WeegeePeer.load(ctx, peer).item)
|
||||
connection.peers = peers
|
||||
if connection.peers:
|
||||
connection.save()
|
||||
else:
|
||||
connection.delete()
|
||||
|
||||
disconnect_peer = peer_commands.add_parser('disconnect', help='disconnect peers')
|
||||
disconnect_peer.add_argument('name', help='connection name')
|
||||
disconnect_peer.add_argument('peer', nargs='+', help='peer names')
|
||||
disconnect_peer.set_defaults(func=do_disconnect_peer, parser=disconnect_peer)
|
||||
|
||||
def do_get_connection(parser: argparse.ArgumentParser, args: argparse.Namespace, ctx: WeegeeContext) -> None:
|
||||
connection = WeegeeConnection.load(ctx, args.name)
|
||||
for var in args.var:
|
||||
do_get(connection, var)
|
||||
|
||||
get_connection = peer_commands.add_parser('get-connect', help='get connection variable')
|
||||
get_connection.add_argument('name', help='connection name')
|
||||
get_connection.add_argument('var', nargs='*', help='variable spec')
|
||||
get_connection.set_defaults(func=do_get_connection, parser=get_connection)
|
||||
add_common_peer_subcommands(peer_commands)
|
||||
|
||||
|
||||
# Server commands
|
||||
|
||||
server = commands.add_parser('server', help='server commands')
|
||||
server.set_defaults(parser=server)
|
||||
server.set_defaults(parser=server, objtype=WeegeeServer)
|
||||
server_commands = server.add_subparsers(title='subcommand')
|
||||
|
||||
def do_add_server(parser: argparse.ArgumentParser, args: argparse.Namespace, ctx: WeegeeContext) -> None:
|
||||
|
|
@ -614,39 +609,13 @@ def main():
|
|||
add_server.add_argument('port', type=int, help='listen port')
|
||||
add_server.set_defaults(func=do_add_server, parser=add_server)
|
||||
|
||||
def do_del_server(parser: argparse.ArgumentParser, args: argparse.Namespace, ctx: WeegeeContext) -> None:
|
||||
if not args.yes_i_want_to_destroy_this:
|
||||
parser.error('please pass --yes-i-want-to-destroy-this if you really want to destroy this server')
|
||||
server = WeegeeServer.load(ctx, args.name)
|
||||
server.delete()
|
||||
|
||||
del_server = server_commands.add_parser('destroy', help='delete server')
|
||||
del_server.add_argument('name', help='server name')
|
||||
del_server.set_defaults(func=do_del_server, parser=del_server)
|
||||
|
||||
def do_conf_server(parser: argparse.ArgumentParser, args: argparse.Namespace, ctx: WeegeeContext) -> None:
|
||||
server = WeegeeServer.load(ctx, args.name)
|
||||
print(server.gen_config())
|
||||
|
||||
conf_server = server_commands.add_parser('print-config', help='show WireGuard configuration')
|
||||
conf_server.add_argument('name', help='server name')
|
||||
conf_server.set_defaults(func=do_conf_server, parser=conf_server)
|
||||
|
||||
def do_get_server(parser: argparse.ArgumentParser, args: argparse.Namespace, ctx: WeegeeContext) -> None:
|
||||
server = WeegeeServer.load(ctx, args.name)
|
||||
for var in args.var:
|
||||
do_get(server, var)
|
||||
|
||||
get_server = server_commands.add_parser('get', help='get server variable')
|
||||
get_server.add_argument('name', help='server name')
|
||||
get_server.add_argument('var', nargs='*', help='variable spec')
|
||||
get_server.set_defaults(func=do_get_server, parser=get_server)
|
||||
add_common_peer_subcommands(server_commands)
|
||||
|
||||
|
||||
# Client commands
|
||||
|
||||
client = commands.add_parser('client', help='client commands')
|
||||
client.set_defaults(parser=client)
|
||||
client.set_defaults(parser=client, objtype=WeegeeClient)
|
||||
client_commands = client.add_subparsers(title='subcommand')
|
||||
|
||||
def do_add_client(parser: argparse.ArgumentParser, args: argparse.Namespace, ctx: WeegeeContext) -> None:
|
||||
|
|
@ -673,39 +642,7 @@ def main():
|
|||
add_client.add_argument('name', help='client name')
|
||||
add_client.set_defaults(func=do_add_client, parser=add_client)
|
||||
|
||||
def do_del_client(parser: argparse.ArgumentParser, args: argparse.Namespace, ctx: WeegeeContext) -> None:
|
||||
if not args.yes_i_want_to_destroy_this:
|
||||
parser.error('please pass --yes-i-want-to-destroy-this if you really want to destroy this client')
|
||||
server = WeegeeServer.load(ctx, args.server)
|
||||
client = server.get_client(args.name)
|
||||
client.delete()
|
||||
|
||||
del_client = client_commands.add_parser('destroy', help='delete client')
|
||||
del_client.add_argument('server', help='server name')
|
||||
del_client.add_argument('name', help='name')
|
||||
del_client.set_defaults(func=do_del_client, parser=del_client)
|
||||
|
||||
def do_conf_client(parser: argparse.ArgumentParser, args: argparse.Namespace, ctx: WeegeeContext) -> None:
|
||||
server = WeegeeServer.load(ctx, args.server)
|
||||
client = server.get_client(args.name)
|
||||
print(client.gen_config())
|
||||
|
||||
conf_client = client_commands.add_parser('print-config', help='show WireGuard configuration')
|
||||
conf_client.add_argument('server', help='server name')
|
||||
conf_client.add_argument('name', help='name')
|
||||
conf_client.set_defaults(func=do_conf_client, parser=conf_client)
|
||||
|
||||
def do_get_client(parser: argparse.ArgumentParser, args: argparse.Namespace, ctx: WeegeeContext) -> None:
|
||||
server = WeegeeServer.load(ctx, args.server)
|
||||
client = server.get_client(args.name)
|
||||
for var in args.var:
|
||||
do_get(client, var)
|
||||
|
||||
get_client = client_commands.add_parser('get', help='get client variable')
|
||||
get_client.add_argument('server', help='server name')
|
||||
get_client.add_argument('name', help='name')
|
||||
get_client.add_argument('var', nargs='*', help='variable spec')
|
||||
get_client.set_defaults(func=do_get_client, parser=get_client)
|
||||
add_common_peer_subcommands(client_commands)
|
||||
|
||||
|
||||
# The main buffet
|
||||
|
|
|
|||
|
|
@ -386,7 +386,6 @@ class WeegeePeer(WeegeeBase):
|
|||
|
||||
@classmethod
|
||||
def create(cls, ctx: WeegeeContext, name: str, interface: WeegeePublicInterface, routes: list[IPNetwork], host: O[str] = None, port: O[int] = None, extra: O[list[str]] = None) -> 'WeegeePeer':
|
||||
print('peer', name)
|
||||
return super().create(ctx, name,
|
||||
interface=interface.item,
|
||||
routes=routes,
|
||||
|
|
|
|||
|
|
@ -28,9 +28,6 @@ class WeegeeServer(WeegeePeer):
|
|||
super().delete(trash=trash)
|
||||
sync_all_interfaces(self.context, auto=True)
|
||||
|
||||
def gen_config(self) -> str:
|
||||
return config_interface(self.interface.to_full())
|
||||
|
||||
@dataclass(eq=False, repr=False)
|
||||
class WeegeeClient(WeegeePeer):
|
||||
@classmethod
|
||||
|
|
@ -53,6 +50,3 @@ class WeegeeClient(WeegeePeer):
|
|||
self.connection.delete(trash=trash)
|
||||
super().delete(trash=trash)
|
||||
sync_all_interfaces(self.context, auto=True)
|
||||
|
||||
def gen_config(self) -> str:
|
||||
return config_interface(self.interface.to_full())
|
||||
|
|
|
|||
Loading…
Reference in New Issue