20440670908

Committed 22 Dec 2025 06:31PM UTC coverage: 88.146% (+0.4%) from 87.727%

Build # 20440670908

Build Type

push

github

Committed by

web-flow

Commit Message

Feat: add traction tenant mode for secure webhook registration (#926)

* feat: implement tenant api fallback for webhook registration

Signed-off-by: Yuki I <omoge.real@gmail.com>

* feat: add traction tenant mode for secure webhook registration

Signed-off-by: Yuki I <omoge.real@gmail.com>

* feat: add traction tenant mode for secure webhook registration

Signed-off-by: Yuki I <omoge.real@gmail.com>

* test: improve coverage for webhook utils and config error handling

Signed-off-by: Yuki I <omoge.real@gmail.com>

* test: fix traction config test failures by setting instance attributes

Signed-off-by: Yuki I <omoge.real@gmail.com>

* feat: unify multi-tenant config and add traction mode support

Signed-off-by: Yuki I <omoge.real@gmail.com>

* feat: unify multi-tenant config and add traction mode support

Signed-off-by: Yuki I <omoge.real@gmail.com>

* refactor: centralize config in .env and clean manage script

Signed-off-by: Yuki I <omoge.real@gmail.com>

* fix: address code review feedback on traction mode

Signed-off-by: Yuki I <omoge.real@gmail.com>

* fix: implement token TTL caching and update deprecation docs

Signed-off-by: Yuki I <omoge.real@gmail.com>

* fix: implement token TTL caching and update deprecation docs

Signed-off-by: Yuki I <omoge.real@gmail.com>

* feat: make token cache TTL configurable via env var

Signed-off-by: Yuki I <omoge.real@gmail.com>

* fix: add validation for token cache TTL

Signed-off-by: Yuki I <omoge.real@gmail.com>

---------

Signed-off-by: Yuki I <omoge.real@gmail.com>

Run Details

127 of 133 new or added lines in 5 files covered. (95.49%)

1 existing line in 1 file now uncovered.

2030 of 2303 relevant lines covered (88.15%)

0.88 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.48

/oidc-controller/api/core/webhook_utils.py

import asyncio
import structlog
import requests
from typing import Callable


logger: structlog.typing.FilteringBoundLogger = structlog.getLogger(__name__)


async def register_tenant_webhook(
    wallet_id: str,
    webhook_url: str,
    admin_url: str,
    api_key: str | None,
    admin_api_key: str | None,
    admin_api_key_name: str | None,
    token_fetcher: Callable[[], str] | None = None,
    use_admin_api: bool = True,
):
    """
    Registers the controller's webhook URL with the ACA-Py Agent Tenant.

    Strategies:
    1. If use_admin_api is True (default for 'multi' mode):
       - Try the Admin API (`/multitenancy/wallet/{id}`).
       - If that fails with 403/404 (Blocked) and token_fetcher is present,
         fallback to the Tenant API (`/tenant/wallet`).

    2. If use_admin_api is False (default for 'traction' mode):
       - Directly fetch token using token_fetcher.
       - Use Tenant API (`/tenant/wallet`) to update webhook.
    """
    if not webhook_url:
        logger.warning(
            "Webhook registration skipped: CONTROLLER_WEB_HOOK_URL is missing. "
            "Verification callbacks may not work."
        )
        return

    if use_admin_api and not wallet_id:
        logger.error("Admin API registration requested but wallet_id is missing.")
        return

    if not webhook_url.startswith("http"):
        logger.error(
            f"Invalid webhook URL format: {webhook_url}. Must start with http:// or https://"
        )
        return

    # Prepare URL with Authentication
    # Ensure API key is in the URL if configured.
    # ACA-Py supports this by appending #key to the URL
    if api_key and "#" not in webhook_url:
        webhook_url = f"{webhook_url}#{api_key}"

    # Security: Mask the API key in logs
    log_safe_url = webhook_url
    if "#" in webhook_url:
        try:
            base, secret = webhook_url.split("#", 1)
            if secret:
                log_safe_url = f"{base}#*****"
        except ValueError:
            pass

    payload = {"wallet_webhook_urls": [webhook_url]}

    max_retries = 5
    base_delay = 2  # seconds

    logger.info(f"Attempting to register webhook: {log_safe_url}")

    for attempt in range(0, max_retries):
        try:
            # STRATEGY 1: Standard Multi-Tenant Admin API
            if use_admin_api:
                headers = {}
                if admin_api_key_name and admin_api_key:
                    headers[admin_api_key_name] = admin_api_key

                target_url = f"{admin_url}/multitenancy/wallet/{wallet_id}"
                logger.debug(f"Attempting Admin API update at {target_url}")

                response = requests.put(
                    target_url, json=payload, headers=headers, timeout=5
                )

                if response.status_code == 200:
                    logger.info(
                        "Successfully registered webhook URL with ACA-Py tenant via Admin API"
                    )
                    return

                # Fallback Logic: If Admin API is blocked (403/404)
                elif response.status_code in [403, 404]:
                    logger.warning(
                        f"Admin API returned {response.status_code}. Checking for Tenant API fallback capability..."
                    )
                    if token_fetcher:
                        if await _register_via_tenant_api(
                            admin_url, payload, token_fetcher
                        ):
                            return
                    else:
                        logger.error(
                            "Cannot fallback to Tenant API: No token fetcher available."
                        )
                        return

                elif response.status_code == 401:
                    logger.error("Admin API Unauthorized (401). Check ADMIN_API_KEY.")
                    return

                elif response.status_code >= 500:
                    logger.warning(
                        f"Webhook registration failed with server error {response.status_code}: {response.text}. Retrying..."
                    )
                else:
                    logger.warning(
                        f"Webhook registration returned unexpected status {response.status_code}: {response.text}"
                    )
                    return

            # STRATEGY 2: Tenant API via Proxy (Traction Mode)
            else:
                if not token_fetcher:
                    logger.error(
                        "Registration via Tenant API proxy requested but no token_fetcher provided."
                    )
                    return

                logger.debug("Attempting Direct Tenant API update")
                if await _register_via_tenant_api(admin_url, payload, token_fetcher):
                    return
                logger.warning("Direct Tenant API update failed. Retrying...")

        except requests.exceptions.ConnectionError:
            logger.warning(f"ACA-Py Agent unreachable at {admin_url}")
        except Exception as e:
            logger.error(f"Unexpected error during webhook registration: {str(e)}")
            return

        if attempt < max_retries - 1:
            delay = base_delay * (2**attempt)
            logger.debug(
                f"Retrying webhook registration in {delay} seconds (Attempt {attempt + 1}/{max_retries})"
            )
            await asyncio.sleep(delay)

    logger.error(
        "Failed to register webhook after multiple attempts. Agent notification may fail."
    )


async def _register_via_tenant_api(
    admin_url: str, payload: dict, token_fetcher: Callable[[], str]
) -> bool:
    """Fallback/Direct: use /tenant/wallet endpoint with provided token fetcher."""
    try:
        # 1. Get Token
        token = token_fetcher()

        if not token:
            logger.error("Tenant Fallback: Token fetcher returned empty token")
            return False

        # 2. Update via Tenant API
        # Using the standard Traction/ACA-Py Tenant endpoint
        tenant_url = f"{admin_url}/tenant/wallet"
        tenant_headers = {"Authorization": f"Bearer {token}"}

        update_res = requests.put(
            tenant_url, json=payload, headers=tenant_headers, timeout=5
        )

        if update_res.status_code == 200:
            logger.info("Successfully registered webhook via Tenant API")
            return True
        elif update_res.status_code >= 500:
            logger.warning(
                f"Tenant API Server Error: {update_res.status_code}. {update_res.text}"
            )
            return False
        else:
            logger.error(
                f"Tenant API Update failed. Status: {update_res.status_code} Body: {update_res.text}"
            )
            return False

    except Exception as e:
        logger.error(f"Tenant API Update Exception: {e}")
        return False

1	import asyncio	1✔
2	import structlog	1✔
3	import requests	1✔
4	from typing import Callable	1✔
5
6
7	logger: structlog.typing.FilteringBoundLogger = structlog.getLogger(__name__)	1✔
8
9
10	async def register_tenant_webhook(	1✔
11	wallet_id: str,
12	webhook_url: str,
13	admin_url: str,
14	api_key: str \| None,
15	admin_api_key: str \| None,
16	admin_api_key_name: str \| None,
17	token_fetcher: Callable[[], str] \| None = None,
18	use_admin_api: bool = True,
19	):
20	"""
21	Registers the controller's webhook URL with the ACA-Py Agent Tenant.
22
23	Strategies:
24	1. If use_admin_api is True (default for 'multi' mode):
25	- Try the Admin API (`/multitenancy/wallet/{id}`).
26	- If that fails with 403/404 (Blocked) and token_fetcher is present,
27	fallback to the Tenant API (`/tenant/wallet`).
28
29	2. If use_admin_api is False (default for 'traction' mode):
30	- Directly fetch token using token_fetcher.
31	- Use Tenant API (`/tenant/wallet`) to update webhook.
32	"""
33	if not webhook_url:	1✔
34	logger.warning(	1✔
35	"Webhook registration skipped: CONTROLLER_WEB_HOOK_URL is missing. "
36	"Verification callbacks may not work."
37	)
38	return	1✔
39
40	if use_admin_api and not wallet_id:	1✔
41	logger.error("Admin API registration requested but wallet_id is missing.")	1✔
42	return	1✔
43
44	if not webhook_url.startswith("http"):	1✔
45	logger.error(	1✔
46	f"Invalid webhook URL format: {webhook_url}. Must start with http:// or https://"
47	)
48	return	1✔
49
50	# Prepare URL with Authentication
51	# Ensure API key is in the URL if configured.
52	# ACA-Py supports this by appending #key to the URL
53	if api_key and "#" not in webhook_url:	1✔
54	webhook_url = f"{webhook_url}#{api_key}"	1✔
55
56	# Security: Mask the API key in logs
57	log_safe_url = webhook_url	1✔
58	if "#" in webhook_url:	1✔
59	try:	1✔
60	base, secret = webhook_url.split("#", 1)	1✔
61	if secret:	1✔
62	log_safe_url = f"{base}#*****"	1✔
NEW 63	except ValueError:	×
NEW 64	pass	×
65
66	payload = {"wallet_webhook_urls": [webhook_url]}	1✔
67
68	max_retries = 5	1✔
69	base_delay = 2 # seconds	1✔
70
71	logger.info(f"Attempting to register webhook: {log_safe_url}")	1✔
72
73	for attempt in range(0, max_retries):	1✔
74	try:	1✔
75	# STRATEGY 1: Standard Multi-Tenant Admin API
76	if use_admin_api:	1✔
77	headers = {}	1✔
78	if admin_api_key_name and admin_api_key:	1✔
79	headers[admin_api_key_name] = admin_api_key	1✔
80
81	target_url = f"{admin_url}/multitenancy/wallet/{wallet_id}"	1✔
82	logger.debug(f"Attempting Admin API update at {target_url}")	1✔
83
84	response = requests.put(	1✔
85	target_url, json=payload, headers=headers, timeout=5
86	)
87
88	if response.status_code == 200:	1✔
89	logger.info(	1✔
90	"Successfully registered webhook URL with ACA-Py tenant via Admin API"
91	)
92	return	1✔
93
94	# Fallback Logic: If Admin API is blocked (403/404)
95	elif response.status_code in [403, 404]:	1✔
96	logger.warning(	1✔
97	f"Admin API returned {response.status_code}. Checking for Tenant API fallback capability..."
98	)
99	if token_fetcher:	1✔
100	if await _register_via_tenant_api(	1✔
101	admin_url, payload, token_fetcher
102	):
103	return	1✔
104	else:
105	logger.error(	1✔
106	"Cannot fallback to Tenant API: No token fetcher available."
107	)
108	return	1✔
109
110	elif response.status_code == 401:	1✔
111	logger.error("Admin API Unauthorized (401). Check ADMIN_API_KEY.")	1✔
112	return	1✔
113
114	elif response.status_code >= 500:	1✔
NEW 115	logger.warning(	×
116	f"Webhook registration failed with server error {response.status_code}: {response.text}. Retrying..."
117	)
118	else:
119	logger.warning(	1✔
120	f"Webhook registration returned unexpected status {response.status_code}: {response.text}"
121	)
122	return	1✔
123
124	# STRATEGY 2: Tenant API via Proxy (Traction Mode)
125	else:
126	if not token_fetcher:	1✔
127	logger.error(	1✔
128	"Registration via Tenant API proxy requested but no token_fetcher provided."
129	)
130	return	1✔
131
132	logger.debug("Attempting Direct Tenant API update")	1✔
133	if await _register_via_tenant_api(admin_url, payload, token_fetcher):	1✔
134	return	1✔
NEW 135	logger.warning("Direct Tenant API update failed. Retrying...")	×
136
137	except requests.exceptions.ConnectionError:	1✔
138	logger.warning(f"ACA-Py Agent unreachable at {admin_url}")	1✔
139	except Exception as e:	1✔
140	logger.error(f"Unexpected error during webhook registration: {str(e)}")	1✔
141	return	1✔
142
143	if attempt < max_retries - 1:	1✔
144	delay = base_delay * (2**attempt)	1✔
145	logger.debug(	1✔
146	f"Retrying webhook registration in {delay} seconds (Attempt {attempt + 1}/{max_retries})"
147	)
148	await asyncio.sleep(delay)	1✔
149
150	logger.error(	1✔
151	"Failed to register webhook after multiple attempts. Agent notification may fail."
152	)
153
154
155	async def _register_via_tenant_api(	1✔
156	admin_url: str, payload: dict, token_fetcher: Callable[[], str]
157	) -> bool:
158	"""Fallback/Direct: use /tenant/wallet endpoint with provided token fetcher."""
159	try:	1✔
160	# 1. Get Token
161	token = token_fetcher()	1✔
162
163	if not token:	1✔
NEW 164	logger.error("Tenant Fallback: Token fetcher returned empty token")	×
NEW 165	return False	×
166
167	# 2. Update via Tenant API
168	# Using the standard Traction/ACA-Py Tenant endpoint
169	tenant_url = f"{admin_url}/tenant/wallet"	1✔
170	tenant_headers = {"Authorization": f"Bearer {token}"}	1✔
171
172	update_res = requests.put(	1✔
173	tenant_url, json=payload, headers=tenant_headers, timeout=5
174	)
175
176	if update_res.status_code == 200:	1✔
177	logger.info("Successfully registered webhook via Tenant API")	1✔
178	return True	1✔
179	elif update_res.status_code >= 500:	1✔
180	logger.warning(	1✔
181	f"Tenant API Server Error: {update_res.status_code}. {update_res.text}"
182	)
183	return False	1✔
184	else:
185	logger.error(	1✔
186	f"Tenant API Update failed. Status: {update_res.status_code} Body: {update_res.text}"
187	)
188	return False	1✔
189
190	except Exception as e:	1✔
191	logger.error(f"Tenant API Update Exception: {e}")	1✔
192	return False	1✔

openwallet-foundation / acapy-vc-authn-oidc / 20440670908

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous