diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index 3f75586..76a7b07 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -8,6 +8,7 @@ jobs: config: name: "Compute configuration values" uses: ./.github/workflows/config.yml + lint: name: "Lint codebase" uses: plone/meta/.github/workflows/backend-lint.yml@2.x @@ -16,6 +17,8 @@ jobs: with: python-version: ${{ needs.config.outputs.python-version }} plone-version: ${{ needs.config.outputs.plone-version }} + check-typing: true + test: name: "Test codebase" uses: plone/meta/.github/workflows/backend-pytest.yml@2.x diff --git a/Makefile b/Makefile index dbb690c..f94b767 100644 --- a/Makefile +++ b/Makefile @@ -138,8 +138,13 @@ format: ## Check and fix code base according to Plone standards @uvx ruff@latest format --config $(BACKEND_FOLDER)/pyproject.toml @uvx zpretty@latest -i src +.PHONY: typecheck +typecheck: ## Run static type checking + @echo "$(GREEN)==> Type check codebase$(RESET)" + @uvx mypy@latest --config-file $(BACKEND_FOLDER)/pyproject.toml src + .PHONY: check -check: format lint ## Check and fix code base according to Plone standards +check: format lint typecheck ## Check and fix code base according to Plone standards ############################################ # i18n diff --git a/README.md b/README.md index 405df1e..e1151ba 100644 --- a/README.md +++ b/README.md @@ -203,6 +203,59 @@ There are some differences in configuration: - Each provider can get an optional entry `propertymap`. It is a mapping from authomatic/provider user properties to plone user properties, like `"fullname": "name",`. Look at each providers documentation which properties are available. +## Migrating user identities + +When migrating a site — for example moving to a new instance — you can export the complete set of user identities stored in the Authomatic plugin to a JSON file and import them again later. + +Two helper functions are available in `pas.plugins.authomatic.utils.exportimport`: + +- `export_plugin_data(path)`: writes all stored identities to the JSON file at `path` and returns the path. +- `import_plugin_data(path)`: reads the JSON file at `path` and restores the identities into the plugin, returning `True` on success (or `False` when no Authomatic plugin is installed). + +Both accept an optional `delimiter` argument used to serialize the provider identity keys (defaults to `|`). +Using the console, to export the identities: + +```python +from pas.plugins.authomatic.utils import exportimport +from pathlib import Path +from plone import api +from zope.component.hooks import setSite + + +path = Path("export_authomatic.json") + +app = globals()["app"] +site = app.Plone +setSite(site) + +with api.env.adopt_roles(["Manager"]): + exportimport.export_plugin_data(path) + +``` + +To import them again: + +```python +from pas.plugins.authomatic.utils import exportimport +from pathlib import Path +from plone import api +from zope.component.hooks import setSite + +import transaction + + +path = Path("export_authomatic.json") + +app = globals()["app"] +site = app.Plone +setSite(site) + +with api.env.adopt_roles(["Manager"]): + exportimport.import_plugin_data(path) + +transaction.commit() +``` + ## Integration with Entra ID Enumeration PAS plugin: if you're using **pas.plugins.authomatic** with *Microsoft Entra ID*, we recommend pairing it with [pas.plugins.eea](https://github.com/eea/pas.plugins.eea) for proper user enumeration and metadata synchronization. This complementary plugin enables listing all the Entra ID users and groups and is compatible with Plone 6.1 and 6.2. diff --git a/news/111.feature b/news/111.feature new file mode 100644 index 0000000..893b8a0 --- /dev/null +++ b/news/111.feature @@ -0,0 +1 @@ +Added type annotations across the package and shipped a `py.typed` marker so downstream code can type-check against it (PEP 561). @ericof diff --git a/news/111.internal b/news/111.internal new file mode 100644 index 0000000..eaf4128 --- /dev/null +++ b/news/111.internal @@ -0,0 +1 @@ +Added a `make typecheck` target (mypy) and wired it into `make check`; modernized `setuphandlers` to use `plone.api` and the `plone.base` `INonInstallable` import. @ericof diff --git a/news/112.internal b/news/112.internal new file mode 100644 index 0000000..6beaa96 --- /dev/null +++ b/news/112.internal @@ -0,0 +1 @@ +Reorganized the package following cookieplone conventions: promoted `utils` to a package (`content`/`plugin`/`request`/`settings`), split `useridfactories` into a package, added `vocabularies` and a `config` module, and slimmed `interfaces` down to interface definitions. Also modernized the `IPloneSiteRoot` and request imports. @ericof diff --git a/news/112.tests b/news/112.tests new file mode 100644 index 0000000..4c8a6fc --- /dev/null +++ b/news/112.tests @@ -0,0 +1 @@ +Expanded test coverage across the add-on: the `utils` subpackages, the user-id factories and their dispatcher, the `vocabularies`, the REST and Zope request adapters, and the `config` validator. @ericof diff --git a/news/113.feature b/news/113.feature new file mode 100644 index 0000000..8996e47 --- /dev/null +++ b/news/113.feature @@ -0,0 +1 @@ +Added `export_plugin_data` and `import_plugin_data` helpers (in `pas.plugins.authomatic.utils.exportimport`) to export the plugin's user identities to a JSON file and import them back, for use during site migrations. @ericof diff --git a/news/114.feature b/news/114.feature new file mode 100644 index 0000000..2c7f753 --- /dev/null +++ b/news/114.feature @@ -0,0 +1 @@ +Changed the default `userid_factory_name` from `uuid` to `username_userid`, so new installations get deterministic, human-meaningful user ids that are easier to target with permissions and group assignments. Existing sites keep their configured value. @ericof diff --git a/pyproject.toml b/pyproject.toml index e3cf357..4c33990 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -43,6 +43,7 @@ dependencies = [ [project.optional-dependencies] test = [ "collective.MockMailHost", + "mypy", "plone.app.robotframework[debug]", "plone.app.testing", "plone.restapi[test]", @@ -190,11 +191,21 @@ preview = true [tool.ruff.lint.per-file-ignores] "tests/*" = ["E501", "RUF001", "S101"] +[tool.mypy] +python_version = "3.11" +ignore_missing_imports = true +# The package uses a PEP 420 native namespace (no __init__.py under src/pas), +# so tell mypy the source root explicitly to avoid "found twice under different +# module names" clashes with the installed ``authomatic`` distribution. +explicit_package_bases = true +namespace_packages = true +mypy_path = "src" + [tool.pytest.ini_options] testpaths = ["tests"] [tool.coverage.run] -source_pkgs = ["pas.plugins.authomatic", "tests"] +source_pkgs = ["pas.plugins.authomatic"] branch = true parallel = true omit = [ diff --git a/src/pas/plugins/authomatic/__init__.py b/src/pas/plugins/authomatic/__init__.py index 9ffc1c3..4853e9b 100644 --- a/src/pas/plugins/authomatic/__init__.py +++ b/src/pas/plugins/authomatic/__init__.py @@ -1,4 +1,5 @@ from pas.plugins.authomatic.patches import apply_patches +from zope.i18nmessageid import MessageFactory import logging @@ -8,6 +9,8 @@ PACKAGE_NAME = "pas.plugins.authomatic" +_ = MessageFactory(PACKAGE_NAME) + logger = logging.getLogger(PACKAGE_NAME) diff --git a/src/pas/plugins/authomatic/_types.py b/src/pas/plugins/authomatic/_types.py new file mode 100644 index 0000000..0c1efc6 --- /dev/null +++ b/src/pas/plugins/authomatic/_types.py @@ -0,0 +1,146 @@ +"""Shared typing helpers for :mod:`pas.plugins.authomatic`. + +This module centralizes the :class:`~typing.TypedDict` definitions used to +document the various mappings handled by the package -- provider +configuration, REST API replies and user enumeration results. +""" + +from typing import Any +from typing import Protocol +from typing import TypedDict + + +class AuthProvider(Protocol): + """Structural type for an ``authomatic`` provider instance. + + ``authomatic`` ships without type information, so we describe here only + the surface consumed by this package. + """ + + name: str + + +class AuthUser(Protocol): + """Structural type for :class:`authomatic.core.User`. + + Only the attributes and methods used by this package are declared. Fields + are optional because ``authomatic`` populates them from provider data. + """ + + id: str | None + username: str | None + name: str | None + email: str | None + data: dict[str, Any] | None + + def to_dict(self) -> dict[str, Any]: ... + + def update(self) -> None: ... + + +class AuthResult(Protocol): + """Structural type for :class:`authomatic.core.LoginResult`.""" + + user: AuthUser + provider: AuthProvider + error: Any + + +class ProviderConfig(TypedDict, total=False): + """Configuration for a single Authomatic provider. + + The values come from the user-provided JSON configuration, so every key + is optional. ``class_`` is stored as a dotted-path string in the JSON but + resolved to the provider class by :func:`.utils.authomatic_cfg`. + """ + + id: int + class_: Any + display: dict[str, Any] + propertymap: dict[str, str | dict[str, str]] + consumer_key: str + consumer_secret: str + access_headers: dict[str, str] + + +#: Mapping of provider name to its :class:`ProviderConfig`. +AuthomaticConfig = dict[str, ProviderConfig] + + +class ErrorDetail(TypedDict): + """Body of an error returned by the REST API services.""" + + type: str + message: str + + +class ErrorReply(TypedDict): + """Error reply envelope returned by the REST API services.""" + + error: ErrorDetail + + +class NextURLReply(TypedDict): + """Successful reply of the ``@login-authomatic`` GET service.""" + + next_url: str + session: str + + +class TokenReply(TypedDict): + """Successful reply of the ``@login-authomatic`` POST service.""" + + token: str + + +class UserInfo(TypedDict): + """User enumeration entry returned by the PAS plugin.""" + + id: str + login: str + pluginid: str + + +class LoginCredentials(TypedDict, total=False): + """Credentials mapping passed to the PAS authentication plugin.""" + + login: str + password: str + + +class LoginProvider(TypedDict): + """A possible login provider.""" + + id: str + plugin: str + title: str + url: str + + +class ProviderButton(TypedDict): + """A provider entry rendered as a login button by the Classic UI view.""" + + identifier: str + title: str + iconclasses: str + buttonclasses: str + as_form: bool + + +class PasPluginsAuthomaticSettings(Protocol): + """Protocol implementation for the IPasPluginsAuthomaticSettings interface.""" + + secret: str + userid_factory_name: str + json_config: str + + +class SerializedPluginData(TypedDict): + """Serialized representation of the PAS plugin state. + + This is used to export and import the plugin configuration and user + identities. + """ + + userid_by_identityinfo: dict[str, str] + useridentities_by_userid: dict[str, dict[str, Any]] diff --git a/src/pas/plugins/authomatic/browser/configure.zcml b/src/pas/plugins/authomatic/browser/configure.zcml index b8b96ee..070c2bb 100644 --- a/src/pas/plugins/authomatic/browser/configure.zcml +++ b/src/pas/plugins/authomatic/browser/configure.zcml @@ -16,7 +16,7 @@ None: super().updateFields() # self.fields['json_config'].widgetFactory = TextLinesFieldWidget - def updateWidgets(self): + def updateWidgets(self) -> None: super().updateWidgets() @@ -26,11 +28,11 @@ class AuthomaticSettingsEditFormSettingsControlPanel( form = AuthomaticSettingsEditForm -@adapter(Interface, IPasPluginsAuthomaticLayer) +@adapter(Interface, ifaces.IPasPluginsAuthomaticLayer) class AuthomaticSettingsConfigletPanel(RegistryConfigletPanel): """Control Panel endpoint""" - schema = IPasPluginsAuthomaticSettings + schema = ifaces.IPasPluginsAuthomaticSettings configlet_id = "authomatic" configlet_category_id = "plone-users" title = _("Authomatic settings") diff --git a/src/pas/plugins/authomatic/browser/view.py b/src/pas/plugins/authomatic/browser/view.py index 9ae5e66..4abd5f4 100644 --- a/src/pas/plugins/authomatic/browser/view.py +++ b/src/pas/plugins/authomatic/browser/view.py @@ -1,55 +1,61 @@ -from authomatic import Authomatic +from __future__ import annotations + +from authomatic.core import Authomatic +from collections.abc import Iterator +from pas.plugins.authomatic import _ +from pas.plugins.authomatic import _types as t from pas.plugins.authomatic import logger +from pas.plugins.authomatic import utils from pas.plugins.authomatic.integration import ZopeRequestAdapter -from pas.plugins.authomatic.interfaces import _ -from pas.plugins.authomatic.utils import authomatic_cfg -from pas.plugins.authomatic.utils import authomatic_settings from plone import api -from plone.base.interfaces.siteroot import INavigationRoot -from plone.protect.interfaces import IDisableCSRFProtection -from Products.CMFCore.interfaces import ISiteRoot from Products.Five.browser import BrowserView from Products.Five.browser.pagetemplatefile import ViewPageTemplateFile -from zope.interface import alsoProvides +from Products.PluggableAuthService.PluggableAuthService import PluggableAuthService +from typing import cast from zope.interface import implementer from zope.publisher.interfaces import IPublishTraverse - - -def is_root(obj): - """Check if current context is Navigation root or a Portal.""" - return ISiteRoot.providedBy(obj) or INavigationRoot.providedBy(obj) +from ZPublisher.HTTPRequest import WSGIRequest @implementer(IPublishTraverse) class AuthomaticView(BrowserView): template = ViewPageTemplateFile("authomatic.pt") + _config: t.AuthomaticConfig zope_request_adapter_factory = ZopeRequestAdapter @property - def zope_request_adapter(self): + def zope_request_adapter(self) -> ZopeRequestAdapter: return self.zope_request_adapter_factory(self) - def publishTraverse(self, request, name): + def publishTraverse(self, request: WSGIRequest, name: str) -> AuthomaticView: if name and not hasattr(self, "provider"): self.provider = name return self @property - def _provider_names(self): - cfgs = authomatic_cfg() - if not cfgs: + def aclu(self) -> PluggableAuthService: + return api.portal.get_tool("acl_users") + + @property + def config(self) -> t.AuthomaticConfig: + if not hasattr(self, "_config"): + self._config = utils.authomatic_cfg() + return self._config + + @property + def _provider_names(self) -> list[str]: + if not (cfgs := self.config): raise ValueError("Authomatic configuration has errors.") return list(cfgs.keys()) - def providers(self): - cfgs = authomatic_cfg() - if not cfgs: + def providers(self) -> Iterator[t.ProviderButton]: + if not (cfgs := self.config): raise ValueError("Authomatic configuration has errors.") for identifier, cfg in cfgs.items(): entry = cfg.get("display", {}) cssclasses = entry.get("cssclasses", {}) - record = { + record: t.ProviderButton = { "identifier": identifier, "title": entry.get("title", identifier), "iconclasses": cssclasses.get("icon", "glypicon glyphicon-log-in"), @@ -60,10 +66,10 @@ def providers(self): } yield record - def _add_identity(self, result, provider_name): + def _add_identity(self, result: t.AuthResult, provider_name: str) -> None: # delegate to PAS plugin to add the identity - alsoProvides(self.request, IDisableCSRFProtection) - aclu = api.portal.get_tool("acl_users") + utils.disable_csrf_protection(self.request) + aclu = self.aclu aclu.authomatic.remember_identity(result) api.portal.show_message( _( @@ -74,9 +80,9 @@ def _add_identity(self, result, provider_name): self.request, ) - def _remember_identity(self, result, provider_name): - alsoProvides(self.request, IDisableCSRFProtection) - aclu = api.portal.get_tool("acl_users") + def _remember_identity(self, result: t.AuthResult, provider_name: str) -> None: + utils.disable_csrf_protection(self.request) + aclu = self.aclu aclu.authomatic.remember(result) api.portal.show_message( _( @@ -87,17 +93,27 @@ def _remember_identity(self, result, provider_name): self.request, ) - def _handle_error(self, error): + def _handle_error(self, error) -> str: try: return error.message except AttributeError: return str(error) - def __call__(self): + def _redirect(self) -> str: + next_url = self.request.cookies.get("next_url", "") + self.request.response.expireCookie("next_url") + self.request.response.redirect(self.context.absolute_url() + next_url) + return _("redirecting") + + @property + def is_anon(self) -> bool: + return api.user.is_anonymous() + + def __call__(self) -> str | None: provider = getattr(self, "provider", "") - if (cfg := authomatic_cfg()) is None: + if not (cfg := utils.authomatic_cfg()): return _("Authomatic is not configured") - if not is_root(self.context): + if not utils.is_root(self.context): # callback url is expected on either navigationroot or site root # so bevor going on redirect root = api.portal.get_navigation_root(self.context) @@ -117,13 +133,15 @@ def __call__(self): # TODO: some sort of CSRF check might be needed, so that # not an account got connected by CSRF. Research needed. pass - secret = authomatic_settings().secret + secret = utils.authomatic_settings().secret auth = Authomatic(cfg, secret=secret) - result = auth.login(self.zope_request_adapter, self.provider) + result = cast( + "t.AuthResult | None", auth.login(self.zope_request_adapter, self.provider) + ) if not result: logger.info("return from view") # let authomatic do its work - return + return None elif error := result.error: return self._handle_error(error) display = cfg[self.provider].get("display", {}) @@ -136,13 +154,3 @@ def __call__(self): self._remember_identity(result, provider_name) return self._redirect() - - def _redirect(self): - next_url = self.request.cookies.get("next_url", "") - self.request.response.expireCookie("next_url") - self.request.response.redirect(self.context.absolute_url() + next_url) - return _("redirecting") - - @property - def is_anon(self): - return api.user.is_anonymous() diff --git a/src/pas/plugins/authomatic/config.py b/src/pas/plugins/authomatic/config.py new file mode 100644 index 0000000..c161fc8 --- /dev/null +++ b/src/pas/plugins/authomatic/config.py @@ -0,0 +1,78 @@ +"""Package constants and configuration helpers. + +This is a dependency-light leaf module: it must not import from +:mod:`pas.plugins.authomatic.interfaces` (or anything that imports it) so it +can be safely imported by both ``interfaces`` and ``utils``. +""" + +from pas.plugins.authomatic import _ +from zope.interface import Invalid + +import json +import random +import string + + +DEFAULT_ID = "authomatic" + +DEFAULT_CONFIG = """\ +{ + "github": { + "id": 1, + "display": { + "title": "Github", + "cssclasses": { + "button": "plone-btn plone-btn-default", + "icon": "glypicon glyphicon-github" + }, + "as_form": false + }, + "propertymap": { + "email": "email", + "link": "home_page", + "location": "location", + "name": "fullname" + }, + "class_": "authomatic.providers.oauth2.GitHub", + "consumer_key": "Example, please get a key and secret. See", + "consumer_secret": "https://github.com/settings/applications/new", + "access_headers": { + "User-Agent": "Plone (pas.plugins.authomatic)" + } + } +} +""" + +random_secret = "".join( + random.SystemRandom().choice(string.ascii_letters + string.digits) + for _ in range(10) +) + + +def validate_cfg_json(value: str) -> bool: + """Check that we have at least valid json and it is a dict. + + :param value: JSON configuration to validate. + :returns: ``True`` when the configuration is valid. + :raises Invalid: when the configuration is not a non-empty JSON mapping. + """ + try: + jv = json.loads(value) + except json.JSONDecodeError as e: + raise Invalid( + _( + "invalid_json", + "JSON is not valid, parser complained: ${message}", + mapping={"message": f"{e.msg} {e.pos}"}, + ) + ) from None + if not isinstance(jv, dict): + raise Invalid(_("invalid_cfg_no_dict", "JSON root must be a mapping (dict)")) + if len(jv) < 1: + raise Invalid( + _( + "invalid_cfg_empty_dict", + "At least one provider must be configured.", + ) + ) + return True diff --git a/src/pas/plugins/authomatic/configure.zcml b/src/pas/plugins/authomatic/configure.zcml index 76f0f43..d75c0fb 100644 --- a/src/pas/plugins/authomatic/configure.zcml +++ b/src/pas/plugins/authomatic/configure.zcml @@ -1,16 +1,10 @@ - - + + + + + - - - - - diff --git a/src/pas/plugins/authomatic/dependencies.zcml b/src/pas/plugins/authomatic/dependencies.zcml new file mode 100644 index 0000000..643a395 --- /dev/null +++ b/src/pas/plugins/authomatic/dependencies.zcml @@ -0,0 +1,10 @@ + + + + + + + diff --git a/src/pas/plugins/authomatic/integration/restapi.py b/src/pas/plugins/authomatic/integration/restapi.py index 293c92b..1d9a881 100644 --- a/src/pas/plugins/authomatic/integration/restapi.py +++ b/src/pas/plugins/authomatic/integration/restapi.py @@ -1,5 +1,13 @@ +from __future__ import annotations + from authomatic.adapters import BaseAdapter from pas.plugins.authomatic import logger +from pas.plugins.authomatic import utils +from typing import TYPE_CHECKING + + +if TYPE_CHECKING: + from pas.plugins.authomatic.services.authomatic import LoginAuthomatic Headers = dict | None @@ -12,8 +20,12 @@ class RestAPIAdapter(BaseAdapter): frontend_route: str = "login-authomatic" def __init__( - self, view, provider: str, params: Headers = None, cookies: Headers = None - ): + self, + view: LoginAuthomatic, + provider: str, + params: Headers = None, + cookies: Headers = None, + ) -> None: """Initialize the adapter. :param view: Service @@ -41,17 +53,12 @@ def url(self) -> str: return f"{self.public_url}/{self.frontend_route}/{self.provider}" @property - def params(self): + def params(self) -> dict: """HTTP parameters (GET/POST). :returns: Dictionary with HTTP parameters. """ - params = self._params - if not params: - params = dict(self.view.request.form) - to_remove = ["provider", "publicUrl"] - params = {k: v for k, v in params.items() if k not in to_remove} - return params + return self._params or utils.extract_adapter_params(self.view.request) @property def cookies(self) -> dict: @@ -65,11 +72,11 @@ def cookies(self) -> dict: # Response # ========================================================================= - def write(self, value: str): + def write(self, value: str) -> None: """Log Authomatic attempts to write to response.""" logger.debug(f"Authomatic wrote {value} to response.") - def set_header(self, key: str, value: str): + def set_header(self, key: str, value: str) -> None: """Store Authomatic header values. :params key: Header key. @@ -78,7 +85,7 @@ def set_header(self, key: str, value: str): self.headers[key] = value logger.debug(f"Authomatic set header {key} with {value}to response.") - def set_status(self, status: int): + def set_status(self, status: int) -> None: """Log Authomatic attempts to set status code to response. :param status: Status code. diff --git a/src/pas/plugins/authomatic/integration/zope.py b/src/pas/plugins/authomatic/integration/zope.py index d638137..0c63343 100644 --- a/src/pas/plugins/authomatic/integration/zope.py +++ b/src/pas/plugins/authomatic/integration/zope.py @@ -1,13 +1,20 @@ +from __future__ import annotations + from authomatic.adapters import BaseAdapter from pas.plugins.authomatic import logger +from typing import TYPE_CHECKING + +import http.cookies + -import http +if TYPE_CHECKING: + from pas.plugins.authomatic.browser.view import AuthomaticView class ZopeRequestAdapter(BaseAdapter): """Adapter for Zope2 requests package.""" - def __init__(self, view): + def __init__(self, view: AuthomaticView) -> None: """ :param view: BrowserView @@ -19,18 +26,18 @@ def __init__(self, view): # ========================================================================= @property - def url(self): + def url(self) -> str: view_url = self.view.context.absolute_url() url = f"{view_url}/authomatic-handler/{self.view.provider}" logger.debug("url" + url) return url @property - def params(self): + def params(self) -> dict: return dict(self.view.request.form) @property - def cookies(self): + def cookies(self) -> dict[str, str]: # special handling since zope parsing does to much decoding cookie = http.cookies.SimpleCookie() cookie.load(self.view.request["HTTP_COOKIE"]) @@ -41,16 +48,16 @@ def cookies(self): # Response # ========================================================================= - def write(self, value): + def write(self, value: str) -> None: logger.debug("write " + value) self.view.request.response.write(value) - def set_header(self, key, value): + def set_header(self, key: str, value: str) -> None: logger.info("set_header " + key + "=" + value) self.view.request.response.setHeader(key, value) - def set_status(self, status): - code, _ = status.split(" ") - code = int(code) + def set_status(self, status: str) -> None: + raw_code, _ = status.split(" ") + code = int(raw_code) logger.debug(f"set_status {code}") self.view.request.response.setStatus(code) diff --git a/src/pas/plugins/authomatic/interfaces.py b/src/pas/plugins/authomatic/interfaces.py index 548b1d7..603ba32 100644 --- a/src/pas/plugins/authomatic/interfaces.py +++ b/src/pas/plugins/authomatic/interfaces.py @@ -1,88 +1,8 @@ +from pas.plugins.authomatic import _ +from pas.plugins.authomatic import config from zope import schema -from zope.component import getUtilitiesFor -from zope.i18nmessageid import MessageFactory from zope.interface import Interface -from zope.interface import Invalid -from zope.interface import provider from zope.publisher.interfaces.browser import IDefaultBrowserLayer -from zope.schema.interfaces import IVocabularyFactory -from zope.schema.vocabulary import SimpleTerm -from zope.schema.vocabulary import SimpleVocabulary - -import json -import random -import string - - -_ = MessageFactory("pas.plugins.authomatic") - -DEFAULT_ID = "authomatic" - -DEFAULT_CONFIG = """\ -{ - "github": { - "id": 1, - "display": { - "title": "Github", - "cssclasses": { - "button": "plone-btn plone-btn-default", - "icon": "glypicon glyphicon-github" - }, - "as_form": false - }, - "propertymap": { - "email": "email", - "link": "home_page", - "location": "location", - "name": "fullname" - }, - "class_": "authomatic.providers.oauth2.GitHub", - "consumer_key": "Example, please get a key and secret. See", - "consumer_secret": "https://github.com/settings/applications/new", - "access_headers": { - "User-Agent": "Plone (pas.plugins.authomatic)" - } - } -} -""" - -random_secret = "".join( - random.SystemRandom().choice(string.ascii_letters + string.digits) - for _ in range(10) -) - - -def validate_cfg_json(value): - """check that we have at least valid json and its a dict""" - try: - jv = json.loads(value) - except ValueError as e: - raise Invalid( - _( - "invalid_json", - "JSON is not valid, parser complained: ${message}", - mapping={"message": f"{e.msg} {e.pos}"}, - ) - ) from None - if not isinstance(jv, dict): - raise Invalid(_("invalid_cfg_no_dict", "JSON root must be a mapping (dict)")) - if len(jv) < 1: - raise Invalid( - _( - "invalid_cfg_empty_dict", - "At least one provider must be configured.", - ) - ) - return True - - -@provider(IVocabularyFactory) -def userid_factory_vocabulary(context): - items = [] - for name, factory in getUtilitiesFor(IUserIDFactory): - items.append([factory.title, name]) - items = [SimpleTerm(name, name, title) for title, name in sorted(items)] - return SimpleVocabulary(items) class IPasPluginsAuthomaticSettings(Interface): @@ -93,7 +13,7 @@ class IPasPluginsAuthomaticSettings(Interface): default="Some random string used to encrypt the state", ), required=True, - default=random_secret, + default=config.random_secret, ) userid_factory_name = schema.Choice( vocabulary="pas.plugins.authomatic.userid_vocabulary", @@ -104,7 +24,7 @@ class IPasPluginsAuthomaticSettings(Interface): "rare cases in URLs. It is the identifier used for " "the user inside Plone.", ), - default="uuid", + default="username_userid", ) json_config = schema.SourceText( title=_("JSON configuration"), @@ -119,8 +39,8 @@ class IPasPluginsAuthomaticSettings(Interface): '"display" and "propertymap" are special.', ), required=True, - constraint=validate_cfg_json, - default=DEFAULT_CONFIG, + constraint=config.validate_cfg_json, + default=config.DEFAULT_CONFIG, ) diff --git a/src/pas/plugins/authomatic/plugin.py b/src/pas/plugins/authomatic/plugin.py index 571e011..3361c64 100644 --- a/src/pas/plugins/authomatic/plugin.py +++ b/src/pas/plugins/authomatic/plugin.py @@ -1,8 +1,14 @@ +from __future__ import annotations + from AccessControl import ClassSecurityInfo from AccessControl.class_init import InitializeClass from BTrees.OOBTree import OOBTree +from collections.abc import Sequence from operator import itemgetter from pas.plugins.authomatic import logger +from pas.plugins.authomatic._types import AuthResult +from pas.plugins.authomatic._types import LoginCredentials +from pas.plugins.authomatic._types import UserInfo from pas.plugins.authomatic.interfaces import IAuthomaticPlugin from pas.plugins.authomatic.useridentities import UserIdentities from pas.plugins.authomatic.useridfactories import new_userid @@ -15,9 +21,12 @@ from Products.PluggableAuthService.interfaces import plugins as pas_interfaces from Products.PluggableAuthService.interfaces.authservice import _noroles from Products.PluggableAuthService.plugins.BasePlugin import BasePlugin +from Products.PluggableAuthService.UserPropertySheet import UserPropertySheet from Products.PluggableAuthService.utils import createViewName from zope.event import notify from zope.interface import implementer +from ZPublisher.HTTPRequest import WSGIRequest +from ZPublisher.HTTPResponse import HTTPResponse tpl_dir = Path(__file__).parent.resolve() / "browser" @@ -27,11 +36,11 @@ def manage_addAuthomaticPlugin( context, - id, # noQA: A002 - title="", - RESPONSE=None, + id: str, # noQA: A002 + title: str = "", + RESPONSE: HTTPResponse | None = None, **kw, -): +) -> None: """Create an instance of a Authomatic Plugin.""" plugin = AuthomaticPlugin(id, title, **kw) context._setObject(plugin.getId(), plugin) @@ -60,24 +69,25 @@ class AuthomaticPlugin(BasePlugin): security = ClassSecurityInfo() meta_type = "Authomatic Plugin" manage_options = BasePlugin.manage_options + REQUEST: WSGIRequest # Tell PAS not to swallow our exceptions _dont_swallow_my_exceptions = True - def __init__(self, id, title=None, **kw): # noQA: A002 + def __init__(self, id: str, title: str | None = None, **kw) -> None: # noQA: A002 self._setId(id) self.title = title self.plugin_caching = True self._init_trees() - def _init_trees(self): + def _init_trees(self) -> None: # (provider_name, provider_userid) -> userid self._userid_by_identityinfo = OOBTree() # userid -> userdata self._useridentities_by_userid = OOBTree() - def _provider_id(self, result): + def _provider_id(self, result: AuthResult) -> tuple[str, str]: """helper to get the provider identifier""" if not result.user.id: raise ValueError("Invalid: Empty user.id") @@ -86,7 +96,7 @@ def _provider_id(self, result): return (result.provider.name, result.user.id) @security.private - def lookup_identities(self, result): + def lookup_identities(self, result: AuthResult) -> UserIdentities | None: """looks up the UserIdentities by using the provider name and the userid at this provider """ @@ -94,7 +104,9 @@ def lookup_identities(self, result): return self._useridentities_by_userid.get(userid, None) @security.private - def remember_identity(self, result, userid=None): + def remember_identity( + self, result: AuthResult, userid: str | None = None + ) -> UserIdentities: """stores authomatic result data""" if userid is None: # create a new userid @@ -114,11 +126,12 @@ def remember_identity(self, result, userid=None): return useridentities @security.private - def remember(self, result): + def remember(self, result: AuthResult) -> None: """remember user as valid result is authomatic result data. """ + request = self.REQUEST # first fetch provider specific user-data result.user.update() @@ -140,7 +153,7 @@ def remember(self, result): aclu = api.portal.get_tool("acl_users") user = aclu._findUser(aclu.plugins, useridentities.userid) accessed, container, name, value = aclu._getObjectContext( - self.REQUEST["PUBLISHED"], self.REQUEST + request["PUBLISHED"], request ) # Add the user to the SM stack aclu._authorizeUser(user, accessed, container, name, value, _noroles) @@ -149,16 +162,18 @@ def remember(self, result): notify(PrincipalCreated(user)) # do login post-processing - self.REQUEST["__ac_password"] = useridentities.secret + request["__ac_password"] = useridentities.secret mt = api.portal.get_tool("portal_membership") logger.info(f"Login Postprocessing: {useridentities.userid}") - mt.loginUser(self.REQUEST) + mt.loginUser(request) # ## # pas_interfaces.IAuthenticationPlugin @security.public - def authenticateCredentials(self, credentials): + def authenticateCredentials( + self, credentials: LoginCredentials + ) -> tuple[str, str] | None: """credentials -> (userid, login) - 'credentials' will be a mapping, as returned by IExtractionPlugin. @@ -173,12 +188,15 @@ def authenticateCredentials(self, credentials): identities = self._useridentities_by_userid[login] if identities.check_password(password): return login, login + return None # ## # pas_interfaces.plugins.IPropertiesPlugin @security.private - def getPropertiesForUser(self, user, request=None): + def getPropertiesForUser( + self, user, request: WSGIRequest | None = None + ) -> UserPropertySheet | None: identity = self._useridentities_by_userid.get(user.getId(), _marker) if identity is _marker: return None @@ -190,13 +208,13 @@ def getPropertiesForUser(self, user, request=None): @security.private def enumerateUsers( self, - id=None, # noQA: A002 - login=None, - exact_match=False, - sort_by=None, - max_results=None, + id: str | None = None, # noQA: A002 + login: str | None = None, + exact_match: bool = False, + sort_by: str | None = None, + max_results: int | None = None, **kw, - ): + ) -> Sequence[UserInfo]: """-> ( user_info_1, ... user_info_N ) o Return mappings for users matching the given criteria. @@ -244,7 +262,7 @@ def enumerateUsers( return () pluginid = self.getId() - ret = [] + ret: list[UserInfo] = [] # shortcut for exact match of login/id identity = None if exact_match and search_id and search_id in self._useridentities_by_userid: @@ -292,29 +310,29 @@ def enumerateUsers( ) @security.public - def allowDeletePrincipal(self, principal_id): + def allowDeletePrincipal(self, principal_id: str) -> bool: """True if this plugin can delete a certain user/group. This is true if this plugin manages the user. """ return principal_id in self._useridentities_by_userid @security.private - def doDeleteUser(self, userid): + def doDeleteUser(self, userid: str) -> None: """Given a user id, delete that user""" return self.removeUser(userid) @security.private - def doChangeUser(self, userid, password=None, **kw): + def doChangeUser(self, userid: str, password: str | None = None, **kw) -> bool: """do nothing""" return False @security.private - def doAddUser(self, login, password): + def doAddUser(self, login: str, password: str) -> bool: """do nothing""" return False @security.private - def getPluginIdByUserId(self, user_id): + def getPluginIdByUserId(self, user_id: str) -> tuple[str, str] | str: """ return the right key for given user_id """ @@ -324,7 +342,7 @@ def getPluginIdByUserId(self, user_id): return "" @security.private - def removeUser(self, user_id): + def removeUser(self, user_id: str) -> None: """ """ # Remove the user from all persistent dicts if user_id not in self._useridentities_by_userid: diff --git a/src/pas/plugins/authomatic/py.typed b/src/pas/plugins/authomatic/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/src/pas/plugins/authomatic/services/authomatic.py b/src/pas/plugins/authomatic/services/authomatic.py index d5cc643..591bd5b 100644 --- a/src/pas/plugins/authomatic/services/authomatic.py +++ b/src/pas/plugins/authomatic/services/authomatic.py @@ -1,18 +1,20 @@ -from authomatic import Authomatic +from __future__ import annotations + +from authomatic.core import Authomatic +from pas.plugins.authomatic import _types as t from pas.plugins.authomatic import logger +from pas.plugins.authomatic import utils from pas.plugins.authomatic.integration import RestAPIAdapter -from pas.plugins.authomatic.utils import authomatic_cfg -from pas.plugins.authomatic.utils import authomatic_settings from plone import api -from plone.protect.interfaces import IDisableCSRFProtection from plone.restapi.deserializer import json_body from plone.restapi.services import Service from Products.PluggableAuthService.interfaces.plugins import IAuthenticationPlugin from transaction.interfaces import NoTransaction +from typing import cast from urllib.parse import parse_qsl -from zope.interface import alsoProvides from zope.interface import implementer from zope.publisher.interfaces import IPublishTraverse +from ZPublisher.HTTPRequest import WSGIRequest import transaction @@ -21,24 +23,25 @@ class LoginAuthomatic(Service): """Base class for Authomatic login.""" + request: WSGIRequest AUTHOMATIC_COOKIE = "authomatic" provider_id: str = "" - _providers = None - _data = None + _providers: t.AuthomaticConfig | None = None + _data: dict | None = None - def publishTraverse(self, request, name): + def publishTraverse(self, request: WSGIRequest, name: str) -> LoginAuthomatic: # Store the first path segment as the provider request["TraversalRequestNameStack"] = [] self.provider_id = name return self @property - def providers(self) -> dict: + def providers(self) -> t.AuthomaticConfig: """Return Authomatic providers.""" providers = self._providers if not providers: try: - providers = authomatic_cfg() + providers = utils.authomatic_cfg() except KeyError: # Authomatic is not configured providers = {} @@ -48,7 +51,7 @@ def providers(self) -> dict: return providers @property - def json_body(self): + def json_body(self) -> dict: if not self._data: self._data = json_body(self.request) return self._data @@ -68,10 +71,10 @@ def public_url(self) -> str: def get_auth(self) -> Authomatic: providers = self.providers - secret = authomatic_settings().secret + secret = utils.authomatic_settings().secret return Authomatic(providers, secret=secret) - def _provider_not_found(self, provider: str) -> dict: + def _provider_not_found(self, provider: str) -> t.ErrorReply: """Return 404 status code for a provider not found.""" self.request.response.setStatus(404) if not provider: @@ -89,7 +92,7 @@ def _provider_not_found(self, provider: str) -> dict: class Get(LoginAuthomatic): """Provide information to start the OAuth process.""" - def extract_cookie_identifier(self, headers: dict) -> str: + def extract_cookie_identifier(self, headers: dict[str, str]) -> str: """Get value of Authomatic cookie. :param headers: Dictionary with headers set by Authomatic. @@ -103,7 +106,7 @@ def extract_cookie_identifier(self, headers: dict) -> str: value = cookie.replace(cookie_prefix, "") return value - def reply(self) -> dict: + def reply(self) -> t.NextURLReply | t.ErrorReply: """Generate URL and session information to be used by the frontend. :returns: URL and session information. @@ -114,7 +117,7 @@ def reply(self) -> dict: auth = self.get_auth() adapter = RestAPIAdapter(self, provider) - result = auth.login(adapter, provider) + result = cast("t.AuthResult | None", auth.login(adapter, provider)) if result and result.error: self.request.response.setStatus(500) return { @@ -162,7 +165,7 @@ def _get_jwt_plugin(self): break return plugin - def _add_identity(self, result, userid=None): + def _add_identity(self, result: t.AuthResult, userid: str | None = None) -> None: """Add an identity to an existing user. :param result: Authomatic login result. @@ -170,7 +173,7 @@ def _add_identity(self, result, userid=None): aclu = self._get_acl_users() aclu.authomatic.remember_identity(result, userid) - def _remember_identity(self, result): + def _remember_identity(self, result: t.AuthResult) -> None: """Store identity information. :param result: Authomatic login result. @@ -191,7 +194,7 @@ def get_token(self, user) -> str: token = plugin.create_token(user.getId(), data=payload) return token - def _annotate_transaction(self, action, user): + def _annotate_transaction(self, action: str, user) -> None: """Add a note to the current transaction.""" try: # Get the current transaction @@ -208,7 +211,7 @@ def _annotate_transaction(self, action, user): msg = f"(Added new identity to user {user_info})" tx.note(msg) - def reply(self) -> dict: + def reply(self) -> t.TokenReply | t.ErrorReply | None: """Process OAuth callback, authenticate the user and return a JWT Token. :returns: Token information. @@ -225,7 +228,7 @@ def reply(self) -> dict: cookies = {self.AUTHOMATIC_COOKIE: data.get("session", "")} adapter = RestAPIAdapter(self, provider, qs, cookies) auth = self.get_auth() - result = auth.login(adapter, provider) + result = cast("t.AuthResult | None", auth.login(adapter, provider)) if result and result.error: self.request.response.setStatus(401) return { @@ -235,7 +238,7 @@ def reply(self) -> dict: } } elif result: - alsoProvides(self.request, IDisableCSRFProtection) + utils.disable_csrf_protection(self.request) action = "" if api.user.is_anonymous(): self._remember_identity(result) @@ -256,3 +259,5 @@ def reply(self) -> dict: if action: self._annotate_transaction(action, user=user) return {"token": self.get_token(user)} + # ``auth.login`` returned no result: nothing to authenticate. + return None diff --git a/src/pas/plugins/authomatic/services/configure.zcml b/src/pas/plugins/authomatic/services/configure.zcml index 319ca49..086a686 100644 --- a/src/pas/plugins/authomatic/services/configure.zcml +++ b/src/pas/plugins/authomatic/services/configure.zcml @@ -14,7 +14,7 @@ @@ -22,7 +22,7 @@ diff --git a/src/pas/plugins/authomatic/services/login.py b/src/pas/plugins/authomatic/services/login.py index 6789301..a41b48a 100644 --- a/src/pas/plugins/authomatic/services/login.py +++ b/src/pas/plugins/authomatic/services/login.py @@ -1,5 +1,7 @@ -from pas.plugins.authomatic.utils import authomatic_cfg +from pas.plugins.authomatic import _types as t +from pas.plugins.authomatic.utils.settings import list_providers from plone.base.interfaces import IPloneSiteRoot +from plone.dexterity.content import DexterityContent from plone.restapi.interfaces import ILoginProviders from zope.component import adapter from zope.interface import implementer @@ -8,28 +10,14 @@ @adapter(IPloneSiteRoot) @implementer(ILoginProviders) class AuthomaticLoginProviders: - def __init__(self, context): + """Adapter returning all configured Authomatic login providers.""" + + def __init__(self, context: DexterityContent) -> None: self.context = context - def get_providers(self) -> list[dict]: + def get_providers(self) -> list[t.LoginProvider]: """List all configured Authomatic plugins. :returns: List of login options. """ - try: - providers = authomatic_cfg() - except KeyError: - # Authomatic is not configured - providers = {} - plugins = [] - for provider_id, provider in providers.items(): - entry = provider.get("display", {}) - title = entry.get("title", provider_id) - - plugins.append({ - "id": provider_id, - "plugin": "authomatic", - "title": title, - "url": f"{self.context.absolute_url()}/@login-oidc/{provider_id}", - }) - return plugins + return list_providers(self.context.absolute_url()) diff --git a/src/pas/plugins/authomatic/setuphandlers.py b/src/pas/plugins/authomatic/setuphandlers.py index bc999af..eab3c47 100644 --- a/src/pas/plugins/authomatic/setuphandlers.py +++ b/src/pas/plugins/authomatic/setuphandlers.py @@ -1,13 +1,16 @@ -from pas.plugins.authomatic.interfaces import DEFAULT_ID +from pas.plugins.authomatic.config import DEFAULT_ID from pas.plugins.authomatic.plugin import AuthomaticPlugin -from Products.CMFPlone.interfaces import INonInstallable +from plone import api +from plone.base.interfaces import INonInstallable +from Products.GenericSetup.tool import SetupTool +from Products.PluggableAuthService.PluggableAuthService import PluggableAuthService from zope.interface import implementer TITLE = "Authomatic OAuth plugin (pas.plugins.authomatic)" -def _add_plugin(pas, pluginid=DEFAULT_ID): +def _add_plugin(pas: PluggableAuthService, pluginid: str = DEFAULT_ID): if pluginid in pas.objectIds(): return f"{TITLE} already installed." if pluginid != DEFAULT_ID: @@ -26,21 +29,25 @@ def _add_plugin(pas, pluginid=DEFAULT_ID): ) -def _remove_plugin(pas, pluginid=DEFAULT_ID): +def _remove_plugin(pas: PluggableAuthService, pluginid: str = DEFAULT_ID): if pluginid in pas.objectIds(): pas.manage_delObjects([pluginid]) -def post_install(context): - _add_plugin(context.aq_parent.acl_users) +def post_install(context: SetupTool): + acl_users = api.portal.get_tool("acl_users") + _add_plugin(acl_users) -def post_uninstall(context): - _remove_plugin(context.aq_parent.acl_users) +def post_uninstall(context: SetupTool): + acl_users = api.portal.get_tool("acl_users") + _remove_plugin(acl_users) @implementer(INonInstallable) class HiddenProfiles: + """Hidden profiles for this package.""" + def getNonInstallableProfiles(self): """Do not show on Plone's list of installable profiles.""" return [ diff --git a/src/pas/plugins/authomatic/useridentities.py b/src/pas/plugins/authomatic/useridentities.py index 7eb9ca4..23ccaf9 100644 --- a/src/pas/plugins/authomatic/useridentities.py +++ b/src/pas/plugins/authomatic/useridentities.py @@ -1,80 +1,129 @@ from authomatic.core import Credentials -from pas.plugins.authomatic import logger +from pas.plugins.authomatic._types import AuthResult +from pas.plugins.authomatic._types import ProviderConfig from pas.plugins.authomatic.utils import authomatic_cfg from persistent import Persistent from persistent.mapping import PersistentMapping from Products.PluggableAuthService.UserPropertySheet import UserPropertySheet +from typing import Any import uuid class UserIdentity(PersistentMapping): - def __init__(self, result): + data: dict + + def __init__(self, result: AuthResult) -> None: super().__init__() self["provider_name"] = result.provider.name self.update(result.user.to_dict()) @property - def credentials(self): + def credentials(self) -> Credentials: cfg = authomatic_cfg() return Credentials.deserialize(cfg, self.user["credentials"]) @credentials.setter - def credentials(self, credentials): + def credentials(self, credentials: Credentials) -> None: self.data["credentials"] = credentials.serialize() + @classmethod + def from_dict(cls, data: dict[str, Any]) -> "UserIdentity": + """Reconstruct a :class:`UserIdentity` from its serialized mapping. + + Bypasses ``__init__`` (which expects an Authomatic result) so an + identity can be rebuilt from exported data. + + :param data: Mapping previously produced by ``dict(identity)``. + :returns: The reconstructed identity. + """ + identity = cls.__new__(cls) + PersistentMapping.__init__(identity) + identity.update(data) + return identity + class UserIdentities(Persistent): - def __init__(self, userid): + userid: str + _identities: PersistentMapping + _sheet: UserPropertySheet | None + _secret: str + + def __init__(self, userid: str) -> None: self.userid = userid self._identities = PersistentMapping() self._sheet = None self._secret = str(uuid.uuid4()) + @classmethod + def from_dict(cls, data: dict[str, Any]) -> "UserIdentities": + """Reconstruct a :class:`UserIdentities` from its serialized form. + + :param data: Mapping with ``userid``, ``secret`` and ``identities`` + keys, as produced during export. + :returns: The reconstructed user identities. + """ + instance = cls(data["userid"]) + instance._secret = data["secret"] + for provider, identity_data in data["identities"].items(): + instance._identities[provider] = UserIdentity.from_dict(identity_data) + return instance + @property - def secret(self): + def secret(self) -> str: return self._secret - def check_password(self, password): + def check_password(self, password: str) -> bool: return password == self._secret - def handle_result(self, result): + def handle_result(self, result: AuthResult) -> None: """add a authomatic result to this user""" self._sheet = None # invalidate property sheet self._identities[result.provider.name] = UserIdentity(result) - def identity(self, provider): + def identity(self, provider: str) -> UserIdentity | None: """users identity at a distinct provider""" return self._identities.get(provider, None) - def update_userdata(self, result): + def update_userdata(self, result: AuthResult) -> None: self._sheet = None # invalidate property sheet identity = self._identities[result.provider.name] identity.update(result.user.to_dict()) + def _properties_from_identity( + self, identity: UserIdentity, cfg: ProviderConfig + ) -> dict[str, Any]: + """return the property for a given identity""" + pdata = {} + for akey, pkey in cfg.get("propertymap", {}).items(): + # Always search first on the user attributes, then on the raw + # data this guaratees we do not break existing configurations + ainfo = identity.get(akey, None) or identity["data"].get(akey, None) + if ainfo is None: + continue + if isinstance(pkey, dict): + for k, v in pkey.items(): + pdata[k] = ainfo.get(v) + else: + pdata[pkey] = ainfo + return pdata + + def _prepare_property_sheet(self) -> dict[str, Any]: + """build a property sheet from the identities""" + pdata = {"id": self.userid} + if cfgs_providers := authomatic_cfg(): + for provider_name, cfg in cfgs_providers.items(): + identity = self.identity(provider_name) + if identity is None: + continue + pdata.update(self._properties_from_identity(identity, cfg)) + return pdata + @property - def propertysheet(self): + def propertysheet(self) -> UserPropertySheet: if self._sheet is not None: return self._sheet # build sheet from identities - pdata = {"id": self.userid} - cfgs_providers = authomatic_cfg() - for provider_name in cfgs_providers: - identity = self.identity(provider_name) - if identity is None: - continue - logger.debug(identity) - cfg = cfgs_providers[provider_name] - for akey, pkey in cfg.get("propertymap", {}).items(): - # Always search first on the user attributes, then on the raw - # data this guaratees we do not break existing configurations - ainfo = identity.get(akey, None) or identity["data"].get(akey, None) - if ainfo is None: - continue - if isinstance(pkey, dict): - for k, v in pkey.items(): - pdata[k] = ainfo.get(v) - else: - pdata[pkey] = ainfo + pdata = self._prepare_property_sheet() self._sheet = UserPropertySheet(**pdata) return self._sheet diff --git a/src/pas/plugins/authomatic/useridfactories.py b/src/pas/plugins/authomatic/useridfactories.py deleted file mode 100644 index ed96247..0000000 --- a/src/pas/plugins/authomatic/useridfactories.py +++ /dev/null @@ -1,57 +0,0 @@ -from pas.plugins.authomatic.interfaces import _ -from pas.plugins.authomatic.interfaces import IUserIDFactory -from pas.plugins.authomatic.utils import authomatic_settings -from zope.component import queryUtility -from zope.interface import implementer - -import uuid - - -@implementer(IUserIDFactory) -class BaseUserIDFactory: - def normalize(self, plugin, result, userid): - new_userid = userid - counter = 2 # first was taken, so logically its second - while new_userid in plugin._useridentities_by_userid: - new_userid = f"{userid}_{counter}" - counter += 1 - return new_userid - - -class UUID4UserIDFactory(BaseUserIDFactory): - title = _("UUID as User ID") - - def __call__(self, plugin, result): - return self.normalize(plugin, result, str(uuid.uuid4())) - - -class ProviderIDUserIDFactory(BaseUserIDFactory): - title = _("Provider User ID") - - def __call__(self, plugin, result): - return self.normalize(plugin, result, result.user.id) - - -class ProviderIDUserNameFactory(BaseUserIDFactory): - title = _("Provider User Name") - - def __call__(self, plugin, result): - return self.normalize(plugin, result, result.user.username) - - -def new_userid(plugin, result): - settings = authomatic_settings() - factory = queryUtility( - IUserIDFactory, name=settings.userid_factory_name, default=UUID4UserIDFactory() - ) - return factory(plugin, result) - - -class ProviderIDUserNameIdFactory(BaseUserIDFactory): - title = _("Provider User Name or User ID") - - def __call__(self, plugin, result): - user_id = result.user.username - if not user_id: - user_id = result.user.id - return self.normalize(plugin, result, user_id) diff --git a/src/pas/plugins/authomatic/useridfactories/__init__.py b/src/pas/plugins/authomatic/useridfactories/__init__.py new file mode 100644 index 0000000..3156625 --- /dev/null +++ b/src/pas/plugins/authomatic/useridfactories/__init__.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +from .userid import ProviderIDUserIDFactory +from .username import ProviderIDUserNameFactory +from .username_userid import ProviderIDUserNameIdFactory +from .uuid_ import UUID4UserIDFactory +from pas.plugins.authomatic._types import AuthResult +from pas.plugins.authomatic.interfaces import IUserIDFactory +from pas.plugins.authomatic.utils import authomatic_settings +from typing import TYPE_CHECKING +from zope.component import queryUtility + + +if TYPE_CHECKING: + from pas.plugins.authomatic.plugin import AuthomaticPlugin + + +def new_userid(plugin: AuthomaticPlugin, result: AuthResult) -> str: + settings = authomatic_settings() + factory = queryUtility( + IUserIDFactory, name=settings.userid_factory_name, default=UUID4UserIDFactory() + ) + return factory(plugin, result) + + +__all__ = ( + "ProviderIDUserIDFactory", + "ProviderIDUserNameFactory", + "ProviderIDUserNameIdFactory", + "UUID4UserIDFactory", + "new_userid", +) diff --git a/src/pas/plugins/authomatic/useridfactories/base.py b/src/pas/plugins/authomatic/useridfactories/base.py new file mode 100644 index 0000000..1a024fe --- /dev/null +++ b/src/pas/plugins/authomatic/useridfactories/base.py @@ -0,0 +1,23 @@ +from __future__ import annotations + +from pas.plugins.authomatic._types import AuthResult +from pas.plugins.authomatic.interfaces import IUserIDFactory +from typing import TYPE_CHECKING +from zope.interface import implementer + + +if TYPE_CHECKING: + from pas.plugins.authomatic.plugin import AuthomaticPlugin + + +@implementer(IUserIDFactory) +class BaseUserIDFactory: + def normalize( + self, plugin: AuthomaticPlugin, result: AuthResult, userid: str + ) -> str: + new_userid = userid + counter = 2 # first was taken, so logically its second + while new_userid in plugin._useridentities_by_userid: + new_userid = f"{userid}_{counter}" + counter += 1 + return new_userid diff --git a/src/pas/plugins/authomatic/useridfactories/configure.zcml b/src/pas/plugins/authomatic/useridfactories/configure.zcml new file mode 100644 index 0000000..4f7fb4f --- /dev/null +++ b/src/pas/plugins/authomatic/useridfactories/configure.zcml @@ -0,0 +1,18 @@ + + + + + + diff --git a/src/pas/plugins/authomatic/useridfactories/userid.py b/src/pas/plugins/authomatic/useridfactories/userid.py new file mode 100644 index 0000000..5fe5725 --- /dev/null +++ b/src/pas/plugins/authomatic/useridfactories/userid.py @@ -0,0 +1,18 @@ +from __future__ import annotations + +from .base import BaseUserIDFactory +from pas.plugins.authomatic import _ +from pas.plugins.authomatic import _types as t +from typing import TYPE_CHECKING + + +if TYPE_CHECKING: + from pas.plugins.authomatic.plugin import AuthomaticPlugin + + +class ProviderIDUserIDFactory(BaseUserIDFactory): + title = _("Provider User ID") + + def __call__(self, plugin: AuthomaticPlugin, result: t.AuthResult) -> str: + user_id = result.user.id or "" + return self.normalize(plugin, result, user_id) diff --git a/src/pas/plugins/authomatic/useridfactories/username.py b/src/pas/plugins/authomatic/useridfactories/username.py new file mode 100644 index 0000000..069e12f --- /dev/null +++ b/src/pas/plugins/authomatic/useridfactories/username.py @@ -0,0 +1,18 @@ +from __future__ import annotations + +from .base import BaseUserIDFactory +from pas.plugins.authomatic import _ +from pas.plugins.authomatic import _types as t +from typing import TYPE_CHECKING + + +if TYPE_CHECKING: + from pas.plugins.authomatic.plugin import AuthomaticPlugin + + +class ProviderIDUserNameFactory(BaseUserIDFactory): + title = _("Provider User Name") + + def __call__(self, plugin: AuthomaticPlugin, result: t.AuthResult) -> str: + username = result.user.username or "" + return self.normalize(plugin, result, username) diff --git a/src/pas/plugins/authomatic/useridfactories/username_userid.py b/src/pas/plugins/authomatic/useridfactories/username_userid.py new file mode 100644 index 0000000..65d9b52 --- /dev/null +++ b/src/pas/plugins/authomatic/useridfactories/username_userid.py @@ -0,0 +1,20 @@ +from __future__ import annotations + +from .base import BaseUserIDFactory +from pas.plugins.authomatic import _ +from pas.plugins.authomatic import _types as t +from typing import TYPE_CHECKING + + +if TYPE_CHECKING: + from pas.plugins.authomatic.plugin import AuthomaticPlugin + + +class ProviderIDUserNameIdFactory(BaseUserIDFactory): + title = _("Provider User Name or User ID") + + def __call__(self, plugin: AuthomaticPlugin, result: t.AuthResult) -> str: + user_id = result.user.username or "" + if not user_id: + user_id = result.user.id or "" + return self.normalize(plugin, result, user_id) diff --git a/src/pas/plugins/authomatic/useridfactories/uuid_.py b/src/pas/plugins/authomatic/useridfactories/uuid_.py new file mode 100644 index 0000000..746adc0 --- /dev/null +++ b/src/pas/plugins/authomatic/useridfactories/uuid_.py @@ -0,0 +1,23 @@ +from __future__ import annotations + +from .base import BaseUserIDFactory +from pas.plugins.authomatic import _ +from pas.plugins.authomatic import _types as t +from typing import TYPE_CHECKING + + +if TYPE_CHECKING: + from pas.plugins.authomatic.plugin import AuthomaticPlugin + +import uuid + + +if TYPE_CHECKING: + from pas.plugins.authomatic.plugin import AuthomaticPlugin + + +class UUID4UserIDFactory(BaseUserIDFactory): + title = _("UUID as User ID") + + def __call__(self, plugin: AuthomaticPlugin, result: t.AuthResult) -> str: + return self.normalize(plugin, result, str(uuid.uuid4())) diff --git a/src/pas/plugins/authomatic/utils.py b/src/pas/plugins/authomatic/utils.py deleted file mode 100644 index 2a445fc..0000000 --- a/src/pas/plugins/authomatic/utils.py +++ /dev/null @@ -1,50 +0,0 @@ -from pas.plugins.authomatic.interfaces import DEFAULT_ID -from pas.plugins.authomatic.interfaces import IPasPluginsAuthomaticSettings -from plone import api -from plone.registry.interfaces import IRegistry -from zope.component import queryUtility -from zope.dottedname.resolve import resolve - -import json - - -def authomatic_plugin(): - """returns the authomatic pas-plugin instance""" - aclu = api.portal.get_tool("acl_users") - # XXX we should better iterate over all plugins and fetch the - # authomatic plugin. There could be even 2 of them, even if this does not - # make sense. - return aclu.get(DEFAULT_ID, None) - - -def authomatic_settings(): - """fetches the authomatic settings from registry""" - registry = queryUtility(IRegistry) - return registry.forInterface(IPasPluginsAuthomaticSettings) - - -def authomatic_cfg(): - """fetches the authomatic configuration from the settings and - returns it as a dict - """ - settings = authomatic_settings() - try: - cfg = json.loads(settings.json_config) - except ValueError: - return None - if not isinstance(cfg, dict): - return None - ids = set() - cnt = 1 - for provider in cfg: - if "class_" in cfg[provider]: - cfg[provider]["class_"] = resolve(cfg[provider]["class_"]) - if "id" in cfg[provider]: - cfg[provider]["id"] = int(cfg[provider]["id"]) - else: - # pick some id - while cnt in ids: - cnt += 1 - cfg[provider]["id"] = cnt - ids.update([cfg[provider]["id"]]) - return cfg diff --git a/src/pas/plugins/authomatic/utils/__init__.py b/src/pas/plugins/authomatic/utils/__init__.py new file mode 100644 index 0000000..e3406a2 --- /dev/null +++ b/src/pas/plugins/authomatic/utils/__init__.py @@ -0,0 +1,16 @@ +from .content import is_root +from .plugin import authomatic_plugin +from .request import disable_csrf_protection +from .request import extract_adapter_params +from .settings import authomatic_cfg +from .settings import authomatic_settings + + +__all__ = [ + "authomatic_cfg", + "authomatic_plugin", + "authomatic_settings", + "disable_csrf_protection", + "extract_adapter_params", + "is_root", +] diff --git a/src/pas/plugins/authomatic/utils/content.py b/src/pas/plugins/authomatic/utils/content.py new file mode 100644 index 0000000..a309437 --- /dev/null +++ b/src/pas/plugins/authomatic/utils/content.py @@ -0,0 +1,8 @@ +from plone.base.interfaces.siteroot import INavigationRoot +from plone.dexterity.content import DexterityContent +from Products.CMFCore.interfaces import ISiteRoot + + +def is_root(obj: DexterityContent) -> bool: + """Check if current context is Navigation root or a Portal.""" + return ISiteRoot.providedBy(obj) or INavigationRoot.providedBy(obj) diff --git a/src/pas/plugins/authomatic/utils/exportimport.py b/src/pas/plugins/authomatic/utils/exportimport.py new file mode 100644 index 0000000..241f931 --- /dev/null +++ b/src/pas/plugins/authomatic/utils/exportimport.py @@ -0,0 +1,108 @@ +from .plugin import authomatic_plugin +from pas.plugins.authomatic import _types as t +from pas.plugins.authomatic.useridentities import UserIdentities +from pathlib import Path +from plone.restapi.serializer.converters import json_compatible +from typing import Any + +import json + + +DEFAULT_DELIMITER = "|" + + +def _useridentities_as_dict(identities: UserIdentities) -> dict[str, Any]: + """Serialize a :class:`UserIdentities` instance to a plain dict. + + :param identities: The user identities to serialize. + :returns: A JSON-serializable mapping of the stored identities. + """ + data = { + "userid": identities.userid, + "secret": identities.secret, + "identities": { + provider: dict(identity) + for provider, identity in identities._identities.items() + }, + } + # No ``context``: identity data is plain values, so the context-free + # ``IJsonCompatible`` path applies. Passing a ``context`` would route to + # ``IContextawareJsonCompatible``, which has no adapter for plain dicts + # and would silently return ``None``. + serialized_data: dict[str, Any] = json_compatible(data) + return serialized_data + + +def _get_plugindata(delimiter: str = DEFAULT_DELIMITER) -> t.SerializedPluginData: + """Serialize the Authomatic plugin state. + + :param delimiter: Separator used to join each ``(provider_name, + provider_id)`` identity-info key into a single string. + :returns: The serialized plugin state (empty when no plugin is available). + """ + userid_by_identityinfo: dict[str, str] = {} + useridentities_by_userid: dict[str, dict[str, Any]] = {} + if plugin := authomatic_plugin(): + userid_by_identityinfo = { + f"{provider_name}{delimiter}{provider_id}": userid + for ( + provider_name, + provider_id, + ), userid in plugin._userid_by_identityinfo.items() + } + useridentities_by_userid = { + userid: _useridentities_as_dict(identities) + for userid, identities in plugin._useridentities_by_userid.items() + } + return { + "userid_by_identityinfo": userid_by_identityinfo, + "useridentities_by_userid": useridentities_by_userid, + } + + +def _set_plugindata( + data: t.SerializedPluginData, delimiter: str = DEFAULT_DELIMITER +) -> bool: + """Restore the Authomatic plugin state from serialized data. + + :param data: Serialized plugin state produced by :func:`_get_plugindata`. + :param delimiter: Separator used to split the identity-info keys back into + ``(provider_name, provider_id)`` tuples. + :returns: ``True`` when the data was imported, ``False`` when no Authomatic + plugin is available. + """ + status = False + if plugin := authomatic_plugin(): + for key, userid in data["userid_by_identityinfo"].items(): + provider_name, provider_id = key.split(delimiter, 1) + plugin._userid_by_identityinfo[(provider_name, provider_id)] = userid + for userid, identities_data in data["useridentities_by_userid"].items(): + plugin._useridentities_by_userid[userid] = UserIdentities.from_dict( + identities_data + ) + status = True + return status + + +def export_plugin_data(path: Path, delimiter: str = DEFAULT_DELIMITER) -> Path: + """Export the plugin's user identities to a JSON file. + + :param path: Destination file for the JSON export. + :param delimiter: Separator used for the identity-info keys. + :returns: The path the data was written to. + """ + data = _get_plugindata(delimiter=delimiter) + path.write_text(json.dumps(data, indent=2)) + return path + + +def import_plugin_data(path: Path, delimiter: str = DEFAULT_DELIMITER) -> bool: + """Import user identities into the plugin from a JSON file. + + :param path: JSON file previously written by :func:`export_plugin_data`. + :param delimiter: Separator used for the identity-info keys. + :returns: ``True`` when the data was imported, ``False`` when no Authomatic + plugin is available. + """ + data = json.loads(path.read_text()) + return _set_plugindata(data, delimiter=delimiter) diff --git a/src/pas/plugins/authomatic/utils/plugin.py b/src/pas/plugins/authomatic/utils/plugin.py new file mode 100644 index 0000000..89ca87d --- /dev/null +++ b/src/pas/plugins/authomatic/utils/plugin.py @@ -0,0 +1,18 @@ +from __future__ import annotations + +from pas.plugins.authomatic.config import DEFAULT_ID +from plone import api +from typing import TYPE_CHECKING + + +if TYPE_CHECKING: + from pas.plugins.authomatic.plugin import AuthomaticPlugin + + +def authomatic_plugin() -> AuthomaticPlugin | None: + """returns the authomatic pas-plugin instance""" + aclu = api.portal.get_tool("acl_users") + # XXX we should better iterate over all plugins and fetch the + # authomatic plugin. There could be even 2 of them, even if this does not + # make sense. + return aclu.get(DEFAULT_ID, None) diff --git a/src/pas/plugins/authomatic/utils/request.py b/src/pas/plugins/authomatic/utils/request.py new file mode 100644 index 0000000..4f216a9 --- /dev/null +++ b/src/pas/plugins/authomatic/utils/request.py @@ -0,0 +1,20 @@ +from plone.protect.interfaces import IDisableCSRFProtection +from typing import Any +from zope.interface import alsoProvides +from ZPublisher.HTTPRequest import WSGIRequest + + +def disable_csrf_protection(request: WSGIRequest) -> None: + """Disable CSRF protection for the given request.""" + alsoProvides(request, IDisableCSRFProtection) + + +def extract_adapter_params(request: WSGIRequest) -> dict[str, Any]: + """Extract adapter parameters from the request. + + :param request: The WSGI request object. + :returns: Dictionary with adapter parameters. + """ + params = dict(request.form) + to_remove = ["provider", "publicUrl"] + return {k: v for k, v in params.items() if k not in to_remove} diff --git a/src/pas/plugins/authomatic/utils/settings.py b/src/pas/plugins/authomatic/utils/settings.py new file mode 100644 index 0000000..64af5e2 --- /dev/null +++ b/src/pas/plugins/authomatic/utils/settings.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +from pas.plugins.authomatic import _types as t +from pas.plugins.authomatic.interfaces import IPasPluginsAuthomaticSettings +from plone.registry.interfaces import IRegistry +from zope.component import queryUtility +from zope.dottedname.resolve import resolve + +import json + + +def authomatic_settings() -> t.PasPluginsAuthomaticSettings: + """fetches the authomatic settings from registry""" + registry = queryUtility(IRegistry) + return registry.forInterface(IPasPluginsAuthomaticSettings) + + +def authomatic_cfg() -> t.AuthomaticConfig: + """fetches the authomatic configuration from the settings and + returns it as a dict + + Returns an empty dict when the configuration is missing or invalid, so + callers can iterate over the result without a ``None`` check. + """ + settings = authomatic_settings() + try: + cfg = json.loads(settings.json_config) + except ValueError: + return {} + if not isinstance(cfg, dict): + return {} + ids = set() + cnt = 1 + for name in cfg: + provider = cfg[name] + if "class_" in provider: + provider["class_"] = resolve(provider["class_"]) + if "id" in provider: + provider["id"] = int(provider["id"]) + else: + # pick some id + while cnt in ids: + cnt += 1 + provider["id"] = cnt + ids.update([provider["id"]]) + return cfg + + +def list_providers(base_url: str) -> list[t.LoginProvider]: + """List all configured Authomatic plugins. + + :returns: List of login options. + """ + providers = authomatic_cfg() + plugins: list[t.LoginProvider] = [] + for provider_id, provider in providers.items(): + entry = provider.get("display", {}) + title = entry.get("title", provider_id) + + plugins.append({ + "id": provider_id, + "plugin": "authomatic", + "title": title, + "url": f"{base_url}/@login-authomatic/{provider_id}", + }) + return plugins diff --git a/src/pas/plugins/authomatic/vocabularies/__init__.py b/src/pas/plugins/authomatic/vocabularies/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/pas/plugins/authomatic/vocabularies/configure.zcml b/src/pas/plugins/authomatic/vocabularies/configure.zcml new file mode 100644 index 0000000..86c1938 --- /dev/null +++ b/src/pas/plugins/authomatic/vocabularies/configure.zcml @@ -0,0 +1,11 @@ + + + + + diff --git a/src/pas/plugins/authomatic/vocabularies/userid.py b/src/pas/plugins/authomatic/vocabularies/userid.py new file mode 100644 index 0000000..fe044b3 --- /dev/null +++ b/src/pas/plugins/authomatic/vocabularies/userid.py @@ -0,0 +1,17 @@ +from pas.plugins.authomatic.interfaces import IUserIDFactory +from plone.dexterity.content import DexterityContent +from zope.component import getUtilitiesFor +from zope.interface import provider +from zope.schema.interfaces import IVocabularyFactory +from zope.schema.vocabulary import SimpleTerm +from zope.schema.vocabulary import SimpleVocabulary + + +@provider(IVocabularyFactory) +def userid_factory_vocabulary(context: DexterityContent) -> SimpleVocabulary: + """Vocabulary of the registered user id factories.""" + items = [] + for name, factory in getUtilitiesFor(IUserIDFactory): + items.append([factory.title, name]) + terms = [SimpleTerm(name, name, title) for title, name in sorted(items)] + return SimpleVocabulary(terms) diff --git a/tests/browser/__init__.py b/tests/browser/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/browser/test_view.py b/tests/browser/test_view.py new file mode 100644 index 0000000..a58b6cd --- /dev/null +++ b/tests/browser/test_view.py @@ -0,0 +1,170 @@ +from pas.plugins.authomatic.browser import view as view_module +from pas.plugins.authomatic.browser.view import AuthomaticView +from plone import api +from plone.app.testing import login +from plone.app.testing import logout +from plone.app.testing import TEST_USER_NAME +from unittest.mock import MagicMock + +import pytest + + +def set_json_config(value: str) -> None: + api.portal.set_registry_record( + "pas.plugins.authomatic.interfaces.IPasPluginsAuthomaticSettings.json_config", + value, + ) + + +class TestAuthomaticView: + @pytest.fixture(autouse=True) + def _setup(self, portal): + self.portal = portal + self.request = portal.REQUEST + self.view = AuthomaticView(portal, self.request) + + def _mock_authomatic(self, monkeypatch, result): + auth = MagicMock() + auth.login.return_value = result + monkeypatch.setattr(view_module, "Authomatic", lambda *a, **kw: auth) + + # -- properties / helpers ------------------------------------------------ + + def test_config_is_cached(self): + assert "github" in self.view.config + # second access returns the cached value + assert self.view.config is self.view._config + + def test_provider_names(self): + assert self.view._provider_names == ["github"] + + def test_provider_names_raises_when_unconfigured(self): + set_json_config("{not valid json") + view = AuthomaticView(self.portal, self.request) + with pytest.raises(ValueError): + _ = view._provider_names + + def test_providers(self): + providers = list(self.view.providers()) + assert len(providers) == 1 + assert providers[0]["identifier"] == "github" + assert providers[0]["title"] == "Github" + assert providers[0]["as_form"] is False + + def test_providers_raises_when_unconfigured(self): + set_json_config("{not valid json") + view = AuthomaticView(self.portal, self.request) + with pytest.raises(ValueError): + list(view.providers()) + + def test_publish_traverse_sets_provider(self): + result = self.view.publishTraverse(self.request, "github") + assert result is self.view + assert self.view.provider == "github" + + def test_publish_traverse_keeps_first_provider(self): + self.view.publishTraverse(self.request, "github") + # A second traversal segment must not override the provider. + self.view.publishTraverse(self.request, "extra") + assert self.view.provider == "github" + + def test_aclu(self): + assert self.view.aclu.getId() == "acl_users" + + def test_zope_request_adapter(self): + from pas.plugins.authomatic.integration import ZopeRequestAdapter + + assert isinstance(self.view.zope_request_adapter, ZopeRequestAdapter) + + def test_is_anon(self): + login(self.portal, TEST_USER_NAME) + assert self.view.is_anon is False + logout() + assert self.view.is_anon is True + + def test_handle_error_with_message(self): + error = MagicMock() + error.message = "boom" + assert self.view._handle_error(error) == "boom" + + def test_handle_error_without_message(self): + assert self.view._handle_error("plain error") == "plain error" + + def test_redirect(self): + assert self.view._redirect() == "redirecting" + + def test_add_identity_delegates_to_plugin(self, monkeypatch): + aclu = MagicMock() + monkeypatch.setattr(AuthomaticView, "aclu", property(lambda self: aclu)) + result = MagicMock() + self.view._add_identity(result, "GitHub") + aclu.authomatic.remember_identity.assert_called_once_with(result) + + def test_remember_identity_delegates_to_plugin(self, monkeypatch): + aclu = MagicMock() + monkeypatch.setattr(AuthomaticView, "aclu", property(lambda self: aclu)) + result = MagicMock() + self.view._remember_identity(result, "GitHub") + aclu.authomatic.remember.assert_called_once_with(result) + + # -- __call__ branches --------------------------------------------------- + + def test_call_not_configured(self): + set_json_config("{not valid json") + assert self.view() == "Authomatic is not configured" + + def test_call_redirects_when_not_root(self): + with api.env.adopt_roles(["Manager"]): + folder = api.content.create( + container=self.portal, type="Folder", id="f1", title="F1" + ) + view = AuthomaticView(folder, self.request) + view.provider = "github" + assert view() == "redirecting" + + def test_call_renders_template_without_provider(self, monkeypatch): + monkeypatch.setattr( + AuthomaticView, "template", lambda self: "TEMPLATE", raising=False + ) + assert self.view() == "TEMPLATE" + + def test_call_provider_not_supported(self): + self.view.provider = "unknown" + assert self.view() == "Provider not supported" + + def test_call_authenticated_and_connected_redirects(self): + # A logged-in user visiting an already-configured provider is + # redirected before the OAuth flow starts. + login(self.portal, TEST_USER_NAME) + self.view.provider = "github" + assert self.view() == "redirecting" + + def test_call_login_returns_none_when_no_result(self, monkeypatch): + logout() + self.view.provider = "github" + self._mock_authomatic(monkeypatch, None) + assert self.view() is None + + def test_call_login_handles_error(self, monkeypatch): + logout() + self.view.provider = "github" + result = MagicMock() + result.error.message = "oauth failed" + self._mock_authomatic(monkeypatch, result) + assert self.view() == "oauth failed" + + def test_call_login_success_anonymous_remembers_identity(self, monkeypatch): + logout() + self.view.provider = "github" + result = MagicMock() + result.error = None + self._mock_authomatic(monkeypatch, result) + calls = {} + monkeypatch.setattr( + AuthomaticView, + "_remember_identity", + lambda self, r, p: calls.setdefault("remember", (r, p)), + ) + monkeypatch.setattr(AuthomaticView, "_redirect", lambda self: "redirected") + assert self.view() == "redirected" + assert calls["remember"][0] is result diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/integration/test_restapi.py b/tests/integration/test_restapi.py new file mode 100644 index 0000000..09354ce --- /dev/null +++ b/tests/integration/test_restapi.py @@ -0,0 +1,50 @@ +from pas.plugins.authomatic.integration.restapi import RestAPIAdapter +from unittest.mock import MagicMock +from zope.publisher.browser import TestRequest + +import pytest + + +@pytest.fixture +def view(): + view = MagicMock() + view.public_url = "http://example.org" + view.request = TestRequest(form={"code": "abc", "provider": "github"}) + return view + + +class TestRestAPIAdapter: + def test_url(self, view): + adapter = RestAPIAdapter(view, "github") + assert adapter.url == "http://example.org/login-authomatic/github" + + def test_params_from_explicit_value(self, view): + adapter = RestAPIAdapter(view, "github", params={"a": "1"}) + assert adapter.params == {"a": "1"} + + def test_params_extracted_from_request(self, view): + # ``provider`` is filtered out by ``extract_adapter_params``. + adapter = RestAPIAdapter(view, "github") + assert adapter.params == {"code": "abc"} + + def test_cookies_default_empty(self, view): + adapter = RestAPIAdapter(view, "github") + assert adapter.cookies == {} + + def test_cookies_from_explicit_value(self, view): + adapter = RestAPIAdapter(view, "github", cookies={"authomatic": "xyz"}) + assert adapter.cookies == {"authomatic": "xyz"} + + def test_write_only_logs(self, view): + # ``write`` must not touch the response, only log. + adapter = RestAPIAdapter(view, "github") + assert adapter.write("payload") is None + + def test_set_header_stores_value(self, view): + adapter = RestAPIAdapter(view, "github") + adapter.set_header("Location", "http://example.org/next") + assert adapter.headers == {"Location": "http://example.org/next"} + + def test_set_status_is_noop(self, view): + adapter = RestAPIAdapter(view, "github") + assert adapter.set_status(302) is None diff --git a/tests/integration/test_zope.py b/tests/integration/test_zope.py new file mode 100644 index 0000000..0327f6c --- /dev/null +++ b/tests/integration/test_zope.py @@ -0,0 +1,47 @@ +from pas.plugins.authomatic.integration.zope import ZopeRequestAdapter +from unittest.mock import MagicMock + +import pytest + + +@pytest.fixture +def view(): + view = MagicMock() + view.context.absolute_url.return_value = "http://example.org/site" + view.provider = "github" + view.request.form = {"code": "abc", "state": "xyz"} + view.request.__getitem__.return_value = "authomatic=cookievalue" + return view + + +class TestZopeRequestAdapter: + def test_url(self, view): + adapter = ZopeRequestAdapter(view) + assert adapter.url == "http://example.org/site/authomatic-handler/github" + + def test_params(self, view): + adapter = ZopeRequestAdapter(view) + assert adapter.params == {"code": "abc", "state": "xyz"} + + def test_cookies(self, view): + # The raw HTTP_COOKIE header is parsed with SimpleCookie. + adapter = ZopeRequestAdapter(view) + assert adapter.cookies == {"authomatic": "cookievalue"} + + def test_write(self, view): + adapter = ZopeRequestAdapter(view) + adapter.write("payload") + view.request.response.write.assert_called_once_with("payload") + + def test_set_header(self, view): + adapter = ZopeRequestAdapter(view) + adapter.set_header("Location", "http://example.org/next") + view.request.response.setHeader.assert_called_once_with( + "Location", "http://example.org/next" + ) + + def test_set_status(self, view): + # Authomatic passes a status line like ``"302 Found"``. + adapter = ZopeRequestAdapter(view) + adapter.set_status("302 Found") + view.request.response.setStatus.assert_called_once_with(302) diff --git a/tests/plugin/test_useridentities.py b/tests/plugin/test_useridentities.py index 9cd1364..2921b89 100644 --- a/tests/plugin/test_useridentities.py +++ b/tests/plugin/test_useridentities.py @@ -144,6 +144,32 @@ def test_read_attribute_from_provider_data_if_default_is_none(self, one_user): sheet = user.propertysheet assert sheet.getProperty("email") == "jdoe@foobar.com" + def test_properties_from_identity_flat_mapping(self, one_user): + user = one_user(self.plugin, self.provider_name, data={}) + identity = user.identity(self.provider_name) + cfg = {"propertymap": {"email": "email"}} + props = user._properties_from_identity(identity, cfg) + assert props == {"email": "andrewpipkin@foobar.com"} + + def test_properties_from_identity_nested_mapping(self, one_user): + # A dict-valued propertymap entry maps sub-keys of the raw provider + # data to distinct Plone properties. + user = one_user( + self.plugin, + self.provider_name, + data={"image": {"url": "https://example.org/a.jpg", "isDefault": False}}, + ) + identity = user.identity(self.provider_name) + cfg = {"propertymap": {"image": {"avatar": "url"}}} + props = user._properties_from_identity(identity, cfg) + assert props == {"avatar": "https://example.org/a.jpg"} + + def test_properties_from_identity_skips_missing_attribute(self, one_user): + user = one_user(self.plugin, self.provider_name, data={}) + identity = user.identity(self.provider_name) + cfg = {"propertymap": {"nonexistent": "whatever"}} + assert user._properties_from_identity(identity, cfg) == {} + class TestUserIdentitiesCustomProps: provider_name: str = "mockhub" diff --git a/tests/plugin/test_useridfactories.py b/tests/plugin/test_useridfactories.py index f811b40..5cdc85e 100644 --- a/tests/plugin/test_useridfactories.py +++ b/tests/plugin/test_useridfactories.py @@ -13,7 +13,7 @@ class MockPlugin: class TestUserIDFactories: def test_normalizer(self, plugin, mock_result): - from pas.plugins.authomatic.useridfactories import BaseUserIDFactory + from pas.plugins.authomatic.useridfactories.base import BaseUserIDFactory bf = BaseUserIDFactory() diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..361e0ad --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,25 @@ +from pas.plugins.authomatic.config import DEFAULT_CONFIG +from pas.plugins.authomatic.config import validate_cfg_json +from zope.interface import Invalid + +import pytest + + +class TestValidateCfgJson: + def test_valid_default_config(self): + assert validate_cfg_json(DEFAULT_CONFIG) is True + + def test_valid_minimal_config(self): + assert validate_cfg_json('{"github": {}}') is True + + def test_invalid_json_raises(self): + with pytest.raises(Invalid): + validate_cfg_json("{not valid json") + + def test_non_mapping_raises(self): + with pytest.raises(Invalid): + validate_cfg_json("[1, 2, 3]") + + def test_empty_mapping_raises(self): + with pytest.raises(Invalid): + validate_cfg_json("{}") diff --git a/tests/useridfactories/__init__.py b/tests/useridfactories/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/useridfactories/conftest.py b/tests/useridfactories/conftest.py new file mode 100644 index 0000000..865e8ac --- /dev/null +++ b/tests/useridfactories/conftest.py @@ -0,0 +1,68 @@ +from authomatic.core import User +from authomatic.providers import oauth2 +from collections.abc import Callable +from dataclasses import dataclass +from pas.plugins.authomatic import _types as t +from pas.plugins.authomatic.plugin import AuthomaticPlugin +from typing import Any +from unittest.mock import MagicMock + +import pytest + + +@dataclass +class LoginResult: + user: User + provider: t.AuthProvider + error: Any = None + + +@pytest.fixture(scope="class") +def plugin() -> AuthomaticPlugin: + + plugin = AuthomaticPlugin(id="authomatic", title="Authomatic") + return plugin + + +@pytest.fixture(scope="session") +def dummy_provider() -> oauth2.OAuth2: + """An OAuth2 provider with mocked settings and adapter. + + The provider is only used to build :class:`~authomatic.core.User` and + login-result instances, so the network-facing ``settings`` and ``adapter`` + collaborators are mocked away. + """ + return oauth2.OAuth2( + settings=MagicMock(), + adapter=MagicMock(), + provider_name="dummy", + ) + + +@pytest.fixture(scope="session") +def auth_user_factory(dummy_provider): + + def factory(user_data: dict[str, Any]) -> User: + user = User(provider=dummy_provider, **user_data) + return user + + return factory + + +@pytest.fixture(scope="session") +def auth_result_factory( + dummy_provider, auth_user_factory +) -> Callable[[dict[str, Any]], t.AuthResult]: + + def factory(user_data: dict[str, Any]) -> LoginResult: + user = auth_user_factory(user_data) + provider = dummy_provider + return LoginResult(user=user, provider=provider) + + return factory + + +@pytest.fixture(scope="class") +def user_data() -> dict[str, Any]: + """Fixture providing sample user data for testing.""" + return {"id": "12345", "username": "testuser", "email": "foo@bar.com"} diff --git a/tests/useridfactories/test_factory_provider_id.py b/tests/useridfactories/test_factory_provider_id.py new file mode 100644 index 0000000..f118e08 --- /dev/null +++ b/tests/useridfactories/test_factory_provider_id.py @@ -0,0 +1,29 @@ +from pas.plugins.authomatic.useridfactories import userid + +import pytest + + +class TestFactory: + factory = userid.ProviderIDUserIDFactory + + @pytest.fixture(autouse=True) + def setup(self, user_data, plugin, auth_result_factory): + self.result = auth_result_factory(user_data) + self.user_id = self.result.user.id + self.plugin = plugin + + def test_factory(self): + """Test that the ProviderIDUserIDFactory generates a valid user ID.""" + factory_instance = self.factory() + user_id = factory_instance(self.plugin, self.result) + assert user_id == self.user_id + + def test_second_call_same_user_id(self): + factory_instance = self.factory() + # Simulate that the user ID is already taken + self.plugin._useridentities_by_userid[self.user_id] = {"Some": "data"} + user_id = factory_instance(self.plugin, self.result) + assert user_id.startswith(self.user_id) + assert ( + user_id == f"{self.user_id}_2" + ) # The second call should append "_2" to the user ID diff --git a/tests/useridfactories/test_factory_username.py b/tests/useridfactories/test_factory_username.py new file mode 100644 index 0000000..331dd19 --- /dev/null +++ b/tests/useridfactories/test_factory_username.py @@ -0,0 +1,29 @@ +from pas.plugins.authomatic.useridfactories import username + +import pytest + + +class TestFactory: + factory = username.ProviderIDUserNameFactory + + @pytest.fixture(autouse=True) + def setup(self, user_data, plugin, auth_result_factory): + self.result = auth_result_factory(user_data) + self.user_id = self.result.user.username + self.plugin = plugin + + def test_factory(self): + """Test that the ProviderIDUserNameFactory generates a valid user ID.""" + factory_instance = self.factory() + user_id = factory_instance(self.plugin, self.result) + assert user_id == self.user_id + + def test_second_call_same_user_id(self): + factory_instance = self.factory() + # Simulate that the user ID is already taken + self.plugin._useridentities_by_userid[self.user_id] = {"Some": "data"} + user_id = factory_instance(self.plugin, self.result) + assert user_id.startswith(self.user_id) + assert ( + user_id == f"{self.user_id}_2" + ) # The second call should append "_2" to the user ID diff --git a/tests/useridfactories/test_factory_username_userid.py b/tests/useridfactories/test_factory_username_userid.py new file mode 100644 index 0000000..a44b19b --- /dev/null +++ b/tests/useridfactories/test_factory_username_userid.py @@ -0,0 +1,34 @@ +from pas.plugins.authomatic.useridfactories import username_userid + +import pytest + + +class TestFactory: + factory = username_userid.ProviderIDUserNameIdFactory + + @pytest.fixture(autouse=True) + def setup(self, user_data, plugin, auth_result_factory): + self.result = auth_result_factory(user_data) + self.plugin = plugin + + def test_factory_uses_username(self): + """When a username is present, it is used as the user ID.""" + factory_instance = self.factory() + user_id = factory_instance(self.plugin, self.result) + assert user_id == self.result.user.username + + def test_factory_falls_back_to_user_id(self, auth_result_factory): + """When the username is empty, the provider user id is used instead.""" + result = auth_result_factory({"id": "12345", "username": "", "email": ""}) + factory_instance = self.factory() + user_id = factory_instance(self.plugin, result) + assert user_id == result.user.id + + def test_second_call_same_user_id(self): + factory_instance = self.factory() + # Simulate that the user ID is already taken + self.plugin._useridentities_by_userid[self.result.user.username] = { + "Some": "data" + } + user_id = factory_instance(self.plugin, self.result) + assert user_id == f"{self.result.user.username}_2" diff --git a/tests/useridfactories/test_factory_uuid4.py b/tests/useridfactories/test_factory_uuid4.py new file mode 100644 index 0000000..e4f108d --- /dev/null +++ b/tests/useridfactories/test_factory_uuid4.py @@ -0,0 +1,28 @@ +from pas.plugins.authomatic.useridfactories import uuid_ +from uuid import UUID + +import pytest + + +class TestFactory: + factory = uuid_.UUID4UserIDFactory + + @pytest.fixture(autouse=True) + def setup(self, user_data, plugin, auth_result_factory): + self.result = auth_result_factory(user_data) + self.plugin = plugin + + def test_factory(self): + """Test that the UUID4UserIDFactory generates a valid UUID4 user ID.""" + factory_instance = self.factory() + user_id = factory_instance(self.plugin, self.result) + + # Check that the generated user ID is a valid UUID4 + try: + uuid_obj = UUID(user_id, version=4) + except ValueError: + pytest.fail(f"Generated user ID '{user_id}' is not a valid UUID4.") + + assert str(uuid_obj) == user_id, ( + "The generated user ID does not match the expected UUID4 format." + ) diff --git a/tests/useridfactories/test_new_userid.py b/tests/useridfactories/test_new_userid.py new file mode 100644 index 0000000..1dc2d81 --- /dev/null +++ b/tests/useridfactories/test_new_userid.py @@ -0,0 +1,40 @@ +from pas.plugins.authomatic.useridfactories import new_userid +from uuid import UUID + +import pytest + + +class TestNewUserID: + """``new_userid`` dispatches to the factory named in the settings. + + This exercises the real registry + utility lookup, so it relies on the + integration ``portal`` fixture (profile installed, factories registered). + """ + + @pytest.fixture(autouse=True) + def setup(self, portal, plugin, auth_result_factory, user_data): + self.portal = portal + self.plugin = plugin + self.result = auth_result_factory(user_data) + + def _set_factory_name(self, name: str): + from plone import api + + api.portal.set_registry_record( + "pas.plugins.authomatic.interfaces." + "IPasPluginsAuthomaticSettings.userid_factory_name", + name, + ) + + def test_dispatches_to_username_factory(self): + self._set_factory_name("username") + assert new_userid(self.plugin, self.result) == self.result.user.username + + def test_dispatches_to_userid_factory(self): + self._set_factory_name("userid") + assert new_userid(self.plugin, self.result) == self.result.user.id + + def test_dispatches_to_uuid_factory(self): + self._set_factory_name("uuid") + user_id = new_userid(self.plugin, self.result) + assert str(UUID(user_id, version=4)) == user_id diff --git a/tests/utils/__init__.py b/tests/utils/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/utils/test_content.py b/tests/utils/test_content.py new file mode 100644 index 0000000..415e3bb --- /dev/null +++ b/tests/utils/test_content.py @@ -0,0 +1,16 @@ +from pas.plugins.authomatic import utils + +import pytest + + +class TestIsRoot: + @pytest.fixture(autouse=True) + def _setup(self, portal_class): + self.portal = portal_class + + def test_portal_is_root(self): + assert utils.is_root(self.portal) is True + + def test_non_root_object(self): + # acl_users provides neither ISiteRoot nor INavigationRoot. + assert utils.is_root(self.portal.acl_users) is False diff --git a/tests/utils/test_exportimport.py b/tests/utils/test_exportimport.py new file mode 100644 index 0000000..cdc3af2 --- /dev/null +++ b/tests/utils/test_exportimport.py @@ -0,0 +1,75 @@ +from pas.plugins.authomatic.useridentities import UserIdentities +from pas.plugins.authomatic.useridentities import UserIdentity +from pas.plugins.authomatic.utils import exportimport + +import pytest + + +IDENTITY = {"provider_name": "github", "id": "42", "email": "a@b.c"} + + +class TestExportImport: + @pytest.fixture(autouse=True) + def _setup(self, portal): + self.portal = portal + self.plugin = self.portal.acl_users["authomatic"] + + def _seed(self): + self.plugin._userid_by_identityinfo[("github", "42")] = "user-1" + identities = UserIdentities("user-1") + identities._secret = "sekret" # noqa: S105 + identities._identities["github"] = UserIdentity.from_dict(IDENTITY) + self.plugin._useridentities_by_userid["user-1"] = identities + + def test_export(self): + self._seed() + data = exportimport._get_plugindata() + assert data["userid_by_identityinfo"] == {"github|42": "user-1"} + assert data["useridentities_by_userid"]["user-1"] == { + "userid": "user-1", + "secret": "sekret", + "identities": {"github": IDENTITY}, + } + + def test_roundtrip(self): + self._seed() + data = exportimport._get_plugindata() + + # Wipe the plugin state, then import it back. + self.plugin._init_trees() + assert exportimport._get_plugindata()["useridentities_by_userid"] == {} + assert exportimport._set_plugindata(data) is True + + assert dict(self.plugin._userid_by_identityinfo) == {("github", "42"): "user-1"} + restored = self.plugin._useridentities_by_userid["user-1"] + assert isinstance(restored, UserIdentities) + assert restored.secret == "sekret" # noqa: S105 + assert dict(restored.identity("github")) == IDENTITY + + def test_get_plugindata_without_plugin(self, monkeypatch): + monkeypatch.setattr(exportimport, "authomatic_plugin", lambda: None) + assert exportimport._get_plugindata() == { + "userid_by_identityinfo": {}, + "useridentities_by_userid": {}, + } + + def test_set_plugindata_without_plugin(self, monkeypatch): + monkeypatch.setattr(exportimport, "authomatic_plugin", lambda: None) + data = {"userid_by_identityinfo": {}, "useridentities_by_userid": {}} + assert exportimport._set_plugindata(data) is False + + def test_export_import_plugin_data_file_roundtrip(self, tmp_path): + self._seed() + path = tmp_path / "identities.json" + + assert exportimport.export_plugin_data(path) == path + assert path.exists() + + # Wipe, then import from the file. + self.plugin._init_trees() + assert exportimport.import_plugin_data(path) is True + + assert dict(self.plugin._userid_by_identityinfo) == {("github", "42"): "user-1"} + restored = self.plugin._useridentities_by_userid["user-1"] + assert restored.secret == "sekret" # noqa: S105 + assert dict(restored.identity("github")) == IDENTITY diff --git a/tests/utils/test_plugin.py b/tests/utils/test_plugin.py new file mode 100644 index 0000000..5cb9c1e --- /dev/null +++ b/tests/utils/test_plugin.py @@ -0,0 +1,16 @@ +from pas.plugins.authomatic import utils + +import pytest + + +class TestAuthomaticPlugin: + @pytest.fixture(autouse=True) + def _setup(self, portal_class): + self.portal = portal_class + + def test_authomatic_plugin_installed(self, plugin_id): + from pas.plugins.authomatic.plugin import AuthomaticPlugin + + plugin = utils.authomatic_plugin() + assert isinstance(plugin, AuthomaticPlugin) + assert plugin.getId() == plugin_id diff --git a/tests/utils/test_request.py b/tests/utils/test_request.py new file mode 100644 index 0000000..71612f9 --- /dev/null +++ b/tests/utils/test_request.py @@ -0,0 +1,33 @@ +from pas.plugins.authomatic import utils +from zope.publisher.browser import TestRequest + + +class TestDisableCSRFProtection: + def test_marks_request(self): + from plone.protect.interfaces import IDisableCSRFProtection + + request = TestRequest() + assert not IDisableCSRFProtection.providedBy(request) + + utils.disable_csrf_protection(request) + assert IDisableCSRFProtection.providedBy(request) + + +class TestExtractAdapterParams: + def test_removes_provider_and_public_url(self): + request = TestRequest( + form={ + "provider": "github", + "publicUrl": "http://example.org", + "code": "abc", + "state": "xyz", + } + ) + assert utils.extract_adapter_params(request) == {"code": "abc", "state": "xyz"} + + def test_empty_form(self): + assert utils.extract_adapter_params(TestRequest()) == {} + + def test_only_filtered_keys_returns_empty(self): + request = TestRequest(form={"provider": "github", "publicUrl": "http://x"}) + assert utils.extract_adapter_params(request) == {} diff --git a/tests/utils/test_settings.py b/tests/utils/test_settings.py new file mode 100644 index 0000000..254ff84 --- /dev/null +++ b/tests/utils/test_settings.py @@ -0,0 +1,83 @@ +from pas.plugins.authomatic import utils +from pas.plugins.authomatic.utils.settings import list_providers + +import pytest + + +def set_json_config(value: str) -> None: + from plone import api + + api.portal.set_registry_record( + "pas.plugins.authomatic.interfaces.IPasPluginsAuthomaticSettings.json_config", + value, + ) + + +class TestSettings: + @pytest.fixture(autouse=True) + def _setup(self, portal_class): + self.portal = portal_class + + def test_authomatic_settings(self): + settings = utils.authomatic_settings() + assert settings.json_config + + def test_default_userid_factory_name(self): + settings = utils.authomatic_settings() + assert settings.userid_factory_name == "username_userid" + + def test_authomatic_cfg_returns_default_provider(self): + cfg = utils.authomatic_cfg() + assert "github" in cfg + # ``class_`` is resolved from its dotted path to the provider class. + assert not isinstance(cfg["github"]["class_"], str) + # ``id`` is coerced to an int. + assert cfg["github"]["id"] == 1 + + def test_authomatic_cfg_invalid_returns_empty_dict(self): + set_json_config("{not valid json") + assert utils.authomatic_cfg() == {} + + def test_authomatic_cfg_non_mapping_returns_empty_dict(self): + set_json_config("[1, 2, 3]") + assert utils.authomatic_cfg() == {} + + def test_authomatic_cfg_assigns_id_when_missing(self): + set_json_config('{"provider_a": {}, "provider_b": {}}') + cfg = utils.authomatic_cfg() + # An id is auto-assigned to every provider lacking one, uniquely. + assert cfg["provider_a"]["id"] == 1 + assert cfg["provider_b"]["id"] == 2 + + +class TestListProviders: + @pytest.fixture(autouse=True) + def _setup(self, portal_class): + self.portal = portal_class + + def test_default_provider(self): + providers = list_providers("http://example.org") + assert providers == [ + { + "id": "github", + "plugin": "authomatic", + "title": "Github", + "url": "http://example.org/@login-authomatic/github", + } + ] + + def test_title_falls_back_to_provider_id(self): + set_json_config('{"acme": {}}') + providers = list_providers("http://example.org") + assert providers == [ + { + "id": "acme", + "plugin": "authomatic", + "title": "acme", + "url": "http://example.org/@login-authomatic/acme", + } + ] + + def test_empty_when_not_configured(self): + set_json_config("{not valid json") + assert list_providers("http://example.org") == [] diff --git a/tests/vocabulary/__init__.py b/tests/vocabulary/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/vocabulary/test_userid_vocabulary.py b/tests/vocabulary/test_userid_vocabulary.py new file mode 100644 index 0000000..47f2f1b --- /dev/null +++ b/tests/vocabulary/test_userid_vocabulary.py @@ -0,0 +1,30 @@ +from plone.app.vocabularies import SimpleVocabulary + +import pytest + + +class TestUserIDVocabulary: + name: str = "pas.plugins.authomatic.userid_vocabulary" + vocab_type = SimpleVocabulary + + @pytest.fixture(autouse=True) + def _setup(self, portal_class, get_vocabulary): + self.portal = portal_class + self.vocab = get_vocabulary(self.name, self.portal) + + def test_vocabulary_type(self): + assert isinstance(self.vocab, self.vocab_type) + + @pytest.mark.parametrize( + "token,title", + [ + ("uuid", "UUID as User ID"), + ("userid", "Provider User ID"), + ("username", "Provider User Name"), + ("username_userid", "Provider User Name or User ID"), + ], + ) + def test_vocab_terms(self, token: str, title: str): + term = self.vocab.getTermByToken(token) + assert term.title == title + assert term.token == token