Merge pull request #99 from icgood/tokens

Use macaroon tokens for admin request auth

README.md

@@ -127,7 +127,7 @@ metadata. It requires [aioredis][9] and will not appear in the plugins list
 without it.
 
 ```
-$ pip install aioredis msgpack
+$ pip install 'pymap[redis]'
 $ pymap redis --help
 ```
 
@@ -164,7 +164,7 @@ instead of a redis hash with the `--users-json` command-line argument.
 Try out the redis plugin:
 
 ```
-$ pymap --port 1143 --debug redis redis://localhost
+$ pymap --port 1143 --debug redis
 ```
 
 Once started, check out the dict plugin example above to connect and see it in
@@ -175,12 +175,12 @@ action.
 The `pymap-admin` tool is installed as an optional dependency:
 
 ```
-$ pip install pymap[admin]
+$ pip install 'pymap[admin]'
 ```
 
-Once installed, subsequent restarts of the pymap server will listen on port
-9090 by default for incoming requests from the admin tool. See the
-[pymap-admin][10] page for more information, and check out the help:
+Once installed, subsequent restarts of the pymap server will listen on a UNIX
+socket for incoming requests from the admin tool. See the [pymap-admin][10]
+page for more information, and check out the help:
 
 ```
 $ pymap-admin --help

doc/source/pymap.admin.rst

@@ -5,6 +5,12 @@
 .. automodule:: pymap.admin
    :members:
 
+.. automodule:: pymap.admin.errors
+   :members:
+
+.. automodule:: pymap.admin.token
+   :members:
+
 ``pymap.admin.handlers``
 ----------------------------
 

doc/source/pymap.interfaces.rst

@@ -7,39 +7,33 @@
 
 .. automodule:: pymap.interfaces.backend
    :members:
-   :special-members: __call__
 
 ``pymap.interfaces.message``
 ----------------------------
 
 .. automodule:: pymap.interfaces.message
    :members:
-   :special-members: __call__
 
 ``pymap.interfaces.mailbox``
 ----------------------------
 
 .. automodule:: pymap.interfaces.mailbox
    :members:
-   :special-members: __call__
 
 ``pymap.interfaces.session``
 ----------------------------
 
 .. automodule:: pymap.interfaces.session
    :members:
-   :special-members: __call__
 
-``pymap.interfaces.users``
+``pymap.interfaces.login``
 --------------------------
 
-.. automodule:: pymap.interfaces.users
+.. automodule:: pymap.interfaces.login
    :members:
-   :special-members: __call__
 
 ``pymap.interfaces.filter``
 ---------------------------
 
 .. automodule:: pymap.interfaces.filter
    :members:
-   :special-members: __call__

pymap/admin/__init__.py

@@ -2,17 +2,23 @@
 from __future__ import annotations
 
 import asyncio
-from argparse import ArgumentParser
+import sys
+from argparse import ArgumentParser, SUPPRESS
 from asyncio import CancelledError
+from datetime import datetime, timedelta, timezone
+from secrets import token_bytes
 from ssl import SSLContext
-from typing import Optional, Tuple, Sequence, Awaitable
-from urllib.parse import urlparse
+from typing import Optional, Sequence, Awaitable
 
+from grpclib.events import listen, RecvRequest
 from grpclib.server import Server
 from pymap.interfaces.backend import BackendInterface, ServiceInterface
+from pymapadmin import is_compatible, __version__ as server_version
 from pymapadmin.local import get_admin_socket
 
+from .errors import get_incompatible_version_error
 from .handlers import handlers
+from .token import get_admin_token
 
 __all__ = ['AdminService']
 
@@ -34,59 +40,69 @@ class AdminService(ServiceInterface):  # pragma: no cover
 
     """
 
+    _admin_token: Optional[bytes] = None
+
     @classmethod
     def add_arguments(cls, parser: ArgumentParser) -> None:
         group = parser.add_argument_group('admin service')
-        group.add_argument('--admin-private', metavar='HOST',
-                           action='append', default=[],
-                           help='private host:port to listen on')
-        group.add_argument('--admin-public', metavar='HOST',
-                           action='append', default=[],
-                           help='public host:port to listen on')
+        group.add_argument('--admin-host', action='append', metavar='IFACE',
+                           help='the network interface to listen on')
+        group.add_argument('--admin-port', metavar='NUM',
+                           help='the port or service name to listen on')
+        group.add_argument('--no-admin-token', action='store_true',
+                           help='disable the admin token')
+        group.add_argument('--admin-token-duration', metavar='SEC', type=int,
+                           help=SUPPRESS)
+
+    def _init_admin_token(self) -> None:
+        if not self.config.args.no_admin_token:
+            expiration: Optional[datetime] = None
+            if self.config.args.admin_token_duration:
+                duration_sec: int = self.config.args.admin_token_duration
+                duration = timedelta(seconds=duration_sec)
+                expiration = datetime.now(tz=timezone.utc) + duration
+            self._admin_token = token_bytes()
+            token = get_admin_token(self._admin_token, expiration)
+            print(f'PYMAP_ADMIN_TOKEN={token.serialize()}', file=sys.stderr)
+
+    async def _check_version(self, event: RecvRequest) -> None:
+        client_version = event.metadata['client-version']
+        if isinstance(client_version, bytes):
+            client_version = client_version.decode('ascii')
+        if not is_compatible(client_version, server_version):
+            raise get_incompatible_version_error(
+                client_version, server_version)
 
     async def start(self) -> Awaitable:
+        self._init_admin_token()
         backend = self.backend
-        private_hosts: Sequence[str] = self.config.args.admin_private
-        public_hosts: Sequence[str] = self.config.args.admin_public
+        host: Optional[str] = self.config.args.admin_host
+        port: Optional[int] = self.config.args.admin_port
         ssl = self.config.ssl_context
         servers = [await self._start_local(backend)]
-        servers += [await self._start(backend, False, host, None)
-                    for host in private_hosts]
-        servers += [await self._start(backend, True, host, ssl)
-                    for host in public_hosts]
+        if host or port:
+            servers += [await self._start(backend, host, port, ssl)]
         return asyncio.create_task(self._run(servers))
 
-    @classmethod
-    def _new_server(cls, backend: BackendInterface, public: bool) -> Server:
-        return Server([handler(backend, public) for _, handler in handlers])
+    def _new_server(self, backend: BackendInterface) -> Server:
+        server = Server([handler(backend, self._admin_token)
+                         for _, handler in handlers])
+        listen(server, RecvRequest, self._check_version)
+        return server
 
-    @classmethod
-    async def _start_local(cls, backend: BackendInterface) -> Server:
-        server = cls._new_server(backend, False)
+    async def _start_local(self, backend: BackendInterface) -> Server:
+        server = self._new_server(backend)
         path = get_admin_socket(mkdir=True)
         await server.start(path=path)
         return server
 
-    @classmethod
-    async def _start(cls, backend: BackendInterface, public: bool,
-                     host: str, ssl: Optional[SSLContext]) -> Server:
-        server = cls._new_server(backend, public)
-        host, port = cls._parse(host)
+    async def _start(self, backend: BackendInterface, host: Optional[str],
+                     port: Optional[int], ssl: Optional[SSLContext]) -> Server:
+        server = self._new_server(backend)
         await server.start(host=host, port=port, ssl=ssl, reuse_address=True)
         return server
 
-    @classmethod
-    def _parse(cls, host: str) -> Tuple[str, Optional[int]]:
-        parsed = urlparse(host)
-        if parsed.hostname:
-            return parsed.hostname, parsed.port
-        parsed = urlparse(f'//{host}')
-        if parsed.hostname:
-            return parsed.hostname, parsed.port
-        raise ValueError(host)
-
-    @classmethod
-    async def _run(cls, servers: Sequence[Server]) -> None:
+    async def _run(self, servers: Sequence[Server]) -> None:
         try:
             for server in servers:
                 await server.wait_closed()

pymap/admin/errors.py

@@ -0,0 +1,42 @@
+
+from __future__ import annotations
+
+from grpclib.const import Status
+from grpclib.exceptions import GRPCError
+from google.rpc.error_details_pb2 import ErrorInfo
+
+__all__ = ['get_unimplemented_error', 'get_incompatible_version_error']
+
+
+def get_unimplemented_error(*, domain: str = None, **metadata: str) \
+        -> GRPCError:
+    """Build a :exc:`~grpclib.exceptions.GRPCError` exception for an
+    operation that is not implemented by the server.
+
+    Args:
+        domain: The domain string to include in the error.
+        metadata: Additional metadata to include in the error.
+
+    """
+    return GRPCError(Status.UNIMPLEMENTED, 'Operation not available', [
+        ErrorInfo(reason='UNIMPLEMENTED', domain=domain, metadata=metadata)])
+
+
+def get_incompatible_version_error(client_version: str, server_version: str, *,
+                                   domain: str = None, **metadata: str) \
+        -> GRPCError:
+    """Build a :exc:`~grpclib.exceptions.GRPCError` exception for an
+    incompatible version error.
+
+    Args:
+        client_version: The client version string.
+        server_version: The server version string.
+        domain: The domain string to include in the error.
+        metadata: Additional metadata to include in the error.
+
+    """
+    metadata = {'client_version': client_version,
+                'server_version': server_version,
+                **metadata}
+    return GRPCError(Status.FAILED_PRECONDITION, 'Incompatible version', [
+        ErrorInfo(reason='INCOMPATIBLE', domain=domain, metadata=metadata)])

pymap/admin/handlers/__init__.py

@@ -3,17 +3,20 @@
 
 from abc import abstractmethod, ABCMeta
 from contextlib import closing, asynccontextmanager, AsyncExitStack
-from typing import Any, Type, Dict, AsyncGenerator
+from typing import Any, Type, Optional, Mapping, Dict, AsyncGenerator
 from typing_extensions import Final
 
 from pymap.context import connection_exit
-from pymap.exceptions import ResponseError
+from pymap.exceptions import InvalidAuth, ResponseError
 from pymap.interfaces.backend import BackendInterface
+from pymap.interfaces.login import IdentityInterface
 from pymap.interfaces.session import SessionInterface
 from pymap.plugin import Plugin
-from pymapadmin.grpc.admin_pb2 import Login, Result, SUCCESS, FAILURE
+from pymapadmin.grpc.admin_pb2 import Result, SUCCESS, FAILURE
 from pysasl import AuthenticationCredentials
-from pysasl.external import ExternalResult
+
+from ..errors import get_unimplemented_error
+from ..token import TokenCredentials
 
 __all__ = ['handlers', 'BaseHandler', 'LoginHandler']
 
@@ -26,15 +29,17 @@ class BaseHandler(metaclass=ABCMeta):
 
     Args:
         backend: The backend in use by the system.
-        public: True if the requests are received from a public-facing client.
+        admin_token: The admin token string that can authenticate any admin
+            operation.
 
     """
 
-    def __init__(self, backend: BackendInterface, public: bool) -> None:
+    def __init__(self, backend: BackendInterface,
+                 admin_token: Optional[bytes]) -> None:
         super().__init__()
         self.config: Final = backend.config
         self.login: Final = backend.login
-        self.public: Final = public
+        self.admin_token: Final = admin_token
 
     @abstractmethod
     def __mapping__(self) -> Dict[str, Any]:
@@ -53,42 +58,50 @@ async def catch_errors(self, command: str) -> AsyncGenerator[Result, None]:
         """
         response = b'. OK %b completed.' % command.encode('utf-8')
         result = Result(code=SUCCESS, response=response)
-        try:
-            yield result
-        except ResponseError as exc:
-            result.code = FAILURE
-            result.response = bytes(exc.get_response(b'.'))
-            result.key = type(exc).__name__
+        async with AsyncExitStack() as stack:
+            connection_exit.set(stack)
+            try:
+                yield result
+            except NotImplementedError as exc:
+                raise get_unimplemented_error() from exc
+            except ResponseError as exc:
+                result.code = FAILURE
+                result.response = bytes(exc.get_response(b'.'))
+                result.key = type(exc).__name__
 
+    @asynccontextmanager
+    async def login_as(self, metadata: Mapping[str, str], user: str) \
+            -> AsyncGenerator[IdentityInterface, None]:
+        """Context manager to login an identity object.
 
-class LoginHandler(BaseHandler, metaclass=ABCMeta):
-    """Base class for implementing admin request handlers that login as a user
-    to handle requests.
+        Args:
+            stream: The grpc request/response stream.
+            user: The user to authorize as.
 
-    Args:
-        backend: The backend in use by the system.
+        Raises:
+            :class:`~pymap.exceptions.InvalidAuth`
 
-    """
+        """
+        try:
+            creds: AuthenticationCredentials = TokenCredentials(
+                metadata['auth-token'], self.admin_token, user)
+        except (KeyError, ValueError) as exc:
+            raise InvalidAuth() from exc
+        yield await self.login.authenticate(creds)
 
     @asynccontextmanager
-    async def login_as(self, login: Login) \
+    async def with_session(self, identity: IdentityInterface) \
             -> AsyncGenerator[SessionInterface, None]:
-        """Context manager to login as a user and provide a session object.
+        """Context manager to create a mail session for the identity.
 
         Args:
-            login: Holds the auuthentication credentials from the request.
+            identity: The authenticated user identity.
 
         Raises:
             :class:`~pymap.exceptions.InvalidAuth`
 
         """
-        if self.public:
-            creds = AuthenticationCredentials(
-                login.authcid, login.secret, login.authzid or None)
-        else:
-            creds = ExternalResult(login.authzid or None)
-        async with AsyncExitStack() as stack:
-            connection_exit.set(stack)
-            session = await stack.enter_async_context(self.login(creds))
-            stack.enter_context(closing(session))
-            yield session
+        stack = connection_exit.get()
+        session = await stack.enter_async_context(identity.new_session())
+        stack.enter_context(closing(session))
+        yield session