from __future__ import annotations
import getpass
from argparse import ArgumentParser, FileType
from contextlib import closing
from typing import Any, TextIO
from .base import Command, AdminCommand
from ..config import Config
from ..local import config_file, token_file
from ..operation import SingleOperation
from ..typing import AdminRequestT, AdminResponseT, MethodProtocol
from ..grpc.admin_grpc import SystemStub
from ..grpc.admin_pb2 import LoginRequest, LoginResponse, \
PingRequest, PingResponse
__all__ = ['SaveArgsCommand', 'LoginCommand', 'PingCommand']
class SystemCommand(AdminCommand[SystemStub, AdminRequestT, AdminResponseT]):
@property
def client(self) -> SystemStub:
return SystemStub(self.channel)
[docs]
class SaveArgsCommand(Command):
"""Save the connection settings given as command-line arguments (e.g.
--host, --port, etc) to a config file.
"""
[docs]
@classmethod
def add_subparser(cls, name: str, subparsers: Any) \
-> ArgumentParser: # pragma: no cover
subparser: ArgumentParser = subparsers.add_parser(
name, description=cls.__doc__,
help='save connection arguments to config file')
return subparser
async def __call__(self, outfile: TextIO, errfile: TextIO) -> int:
path = config_file.get_home(mkdir=True)
parser = Config.build(self.args)
with open(path, 'w') as cfg:
parser.write(cfg)
print('Config file written:', path, file=outfile)
return 0
[docs]
class LoginCommand(SystemCommand[LoginRequest, LoginResponse],
SingleOperation[LoginRequest, LoginResponse]):
"""Login as a user for future requests."""
[docs]
@classmethod
def add_subparser(cls, name: str, subparsers: Any) \
-> ArgumentParser: # pragma: no cover
subparser: ArgumentParser = subparsers.add_parser(
name, description=cls.__doc__,
help='login as a user')
subparser.add_argument('-s', '--save', action='store_true',
help='save the login token')
subparser.add_argument('-z', '--authzid', metavar='NAME',
help='authorization identity name')
subparser.add_argument('--expiration', metavar='TIMESTAMP', type=float,
help='token expiration timestamp')
password = subparser.add_mutually_exclusive_group(required=True)
password.add_argument('-t', '--token', action='store_true',
help='use token authentication to login')
password.add_argument('--password', metavar='VAL',
help='login password')
password.add_argument('--password-file', metavar='PATH',
type=FileType(),
help='file containing login password')
password.add_argument('-i', '--ask-password', action='store_true',
help='read login password from terminal')
subparser.add_argument('user', help='login username')
return subparser
@property
def method(self) -> MethodProtocol[LoginRequest, LoginResponse]:
return self.client.Login
[docs]
def build_request(self) -> LoginRequest:
username: str = self.args.user
authzid: str | None = self.args.authzid
expiration: float | None = self.args.expiration
if self.args.token:
password: str | None = None
elif self.args.ask_password:
password = getpass.getpass(f'{username} Password: ')
elif self.args.password_file is not None:
with closing(self.args.password_file) as pw_file:
password = pw_file.readline().rstrip('\r\n')
else:
password = self.args.password
request = LoginRequest(authcid=username, secret=password)
if authzid is not None:
request.authzid = authzid
if expiration is not None:
request.token_expiration = expiration
return request
[docs]
def handle_success(self, response: LoginResponse,
outfile: TextIO, errfile: TextIO) -> None:
super().handle_success(response, outfile, errfile)
token = response.bearer_token
if token and self.args.save:
path = token_file.get_home(mkdir=True)
path.write_text(token)
path.chmod(0o600)
[docs]
class PingCommand(SystemCommand[PingRequest, PingResponse],
SingleOperation[PingRequest, PingResponse]):
"""Ping the server."""
[docs]
@classmethod
def add_subparser(cls, name: str, subparsers: Any) \
-> ArgumentParser: # pragma: no cover
argparser: ArgumentParser = subparsers.add_parser(
name, description=cls.__doc__,
help='ping the server')
return argparser
@property
def method(self) -> MethodProtocol[PingRequest, PingResponse]:
return self.client.Ping
[docs]
def build_request(self) -> PingRequest:
return PingRequest()