Date: 2026-02-07
Status: Implemented
Components:
factory.core/ObjHook.py (Outbound webhooks)factory.web/WebHooks.py (Inbound webhooks)Webhook credentials (username, password, API keys) can now be stored in Infisical for centralized secret management while maintaining backward compatibility with database storage.
Supports both:
Both implementations use identical patterns for credential lookup and fallback.
When a webhook is executed, credentials are resolved in this order:
webhook.{WEBHOOK_CODE}.{credential_type}def_webhook.Password, def_webhook.User, def_webhook.ApiKey (fallback)# factory.core/ObjHook.py
def _get_webhook_credential(self, credential_type: str, db_value: str) -> str:
"""
Get webhook credential with Infisical integration.
Lookup: webhook.{webhookcode}.{credential_type} → database fallback
"""
section = f"webhook.{self._Webhookcode}"
infisical_value = self.get_ini_value(section, credential_type, "")
if infisical_value:
return infisical_value # From Infisical
return db_value # From database (fallback)
# Automatic Infisical integration (lines 547-550)
self.Username = self._get_webhook_credential("username", self._User)
self.Password = self._get_webhook_credential("password", self._Password)
self.ApiKey = self._get_webhook_credential("apikey", getattr(self, "_ApiKey", ""))
Store credentials in def_webhook table:
UPDATE def_webhook
SET
User = 'api_user',
Password = 'secret_password',
ApiKey = 'sk_live_abc123...'
WHERE Webhookcode = 'STRIPE_PAYMENT';
Store credentials in Infisical secrets manager:
Section: webhook.STRIPE_PAYMENT
| Key | Value | Description |
|---|---|---|
username |
api_user |
API username |
password |
secret_password |
API password |
apikey |
sk_live_abc123... |
API key/token |
# Check Infisical configuration
python factory.core/ConfigIni.py infisical
# Expected output:
# Infisical Status: Connected
# Project: production
# Environment: production
# 1. Extract current credentials from database
python -c "
from factory.core import ObjHook
hook = ObjHook.ObjHook()
hook.read('STRIPE_PAYMENT')
print(f'Username: {hook._User}')
print(f'Password: {hook._Password}')
print(f'ApiKey: {hook._ApiKey}')
"
# 2. Add to Infisical via CLI or Web UI
# Section: webhook.STRIPE_PAYMENT
# Keys: username, password, apikey
# 3. Test webhook with Infisical credentials
python -c "
from factory.core import ObjHook
hook = ObjHook.ObjHook()
result = hook.read('STRIPE_PAYMENT', context={})
print(f'Username from Infisical: {hook.Username}')
print(f'Password from Infisical: {hook.Password}')
"
# 4. Clear database credentials (optional, for security)
python -c "
from factory.core import ObjHook
hook = ObjHook.ObjHook()
hook.sql_execute('''
UPDATE def_webhook
SET User = NULL, Password = NULL, ApiKey = NULL
WHERE Webhookcode = 'STRIPE_PAYMENT'
''')
print('Database credentials cleared')
"
Create resource.bin/migrate_webhooks_to_infisical.py:
#!/usr/bin/env python3
"""
Migrate all webhook credentials from database to Infisical
"""
import sys
import os
base_path = os.getcwd()
sys.path.append(base_path)
sys.path.append(base_path + "/factory.core")
from factory.core.ObjHook import ObjHook
def migrate_webhooks():
"""Migrate all webhooks to Infisical."""
hook = ObjHook()
# Get all webhooks
sql = """
SELECT DISTINCT Webhookcode, User, Password, ApiKey
FROM def_webhook
WHERE Direction = 'OUT'
AND (User IS NOT NULL OR Password IS NOT NULL OR ApiKey IS NOT NULL)
"""
webhooks = hook.sql_get_rows(sql)
print(f"Found {len(webhooks)} webhooks with credentials")
for webhook_code, username, password, apikey in webhooks:
print(f"\nMigrating webhook: {webhook_code}")
# This will automatically attempt to use Infisical
# You need to manually add secrets to Infisical first
print(f" Username: {'SET' if username else 'NOT SET'}")
print(f" Password: {'SET' if password else 'NOT SET'}")
print(f" ApiKey: {'SET' if apikey else 'NOT SET'}")
print(f"\n Add to Infisical:")
print(f" Section: webhook.{webhook_code}")
if username:
print(f" Key: username, Value: {username}")
if password:
print(f" Key: password, Value: ********")
if apikey:
print(f" Key: apikey, Value: ********")
print("\n✅ Migration guide generated")
print(" Manually add secrets to Infisical, then test webhooks")
if __name__ == "__main__":
migrate_webhooks()
# Test webhook reads credentials from Infisical
from factory.core.ObjHook import ObjHook
hook = ObjHook()
hook.read('STRIPE_PAYMENT', context={})
# Check debug output for: "Using password from Infisical for webhook STRIPE_PAYMENT"
print(f"Username: {hook.Username}")
print(f"Password: {'*' * len(hook.Password) if hook.Password else 'NOT SET'}")
print(f"ApiKey: {'*' * len(hook.ApiKey) if hook.ApiKey else 'NOT SET'}")
Infisical Setup:
Section: webhook.STRIPE_PAYMENT
apikey: sk_live_abc123xyz456...
Section: webhook.STRIPE_REFUND
apikey: sk_live_def789ghi012...
Usage:
hook = ObjHook()
hook.read('STRIPE_PAYMENT') # Reads apikey from Infisical
# hook.ApiKey = 'sk_live_abc123xyz456...'
Infisical Setup:
Section: webhook.LEGACY_API
username: api_user_prod
password: complex_password_123!
Usage:
hook = ObjHook()
hook.read('LEGACY_API') # Reads username & password from Infisical
# hook.Username = 'api_user_prod'
# hook.Password = 'complex_password_123!'
Infisical Setup:
Section: webhook.GOOGLE_ANALYTICS
apikey: ya29.a0AfH6SMBx...
Database (deprecated, but still works):
UPDATE def_webhook
SET ApiKey = 'ya29.a0AfH6SMBx...'
WHERE Webhookcode = 'GOOGLE_ANALYTICS';
Stripe webhooks use signing secrets to verify authenticity.
Infisical Setup:
Section: webhook.STRIPE_PAYMENT
apikey: whsec_abc123xyz456...
WebHooks Usage:
from factory.web.WebHooks import WebHooks
webhook = WebHooks()
webhook.Read('STRIPE_PAYMENT')
# webhook._Apikey = 'whsec_abc123xyz456...' (from Infisical)
# Verify Stripe signature (example)
import hmac
import hashlib
def verify_stripe_signature(payload, sig_header, secret):
timestamp, signature = sig_header.split(',')[0].split('=')[1], sig_header.split(',')[1].split('=')[1]
signed_payload = f"{timestamp}.{payload}"
expected_sig = hmac.new(
secret.encode('utf-8'),
signed_payload.encode('utf-8'),
hashlib.sha256
).hexdigest()
return hmac.compare_digest(expected_sig, signature)
# Use the Infisical-loaded secret
is_valid = verify_stripe_signature(
request_body,
request.headers['Stripe-Signature'],
webhook._Apikey # From Infisical
)
GitHub webhooks use HMAC-SHA256 signatures.
Infisical Setup:
Section: webhook.GITHUB_PUSH
apikey: ghp_github_webhook_secret_123
WebHooks Usage:
webhook = WebHooks()
webhook.Read('GITHUB_PUSH')
# Verify GitHub signature
import hmac
import hashlib
def verify_github_signature(payload, signature, secret):
mac = hmac.new(
secret.encode('utf-8'),
payload.encode('utf-8'),
hashlib.sha256
)
return hmac.compare_digest(
'sha256=' + mac.hexdigest(),
signature
)
is_valid = verify_github_signature(
request_body,
request.headers['X-Hub-Signature-256'],
webhook._Apikey # From Infisical
)
Slack uses signing secrets for event subscriptions.
Infisical Setup:
Section: webhook.SLACK_EVENTS
apikey: 8f742231b10e8888abcd1234
WebHooks Usage:
webhook = WebHooks()
webhook.Read('SLACK_EVENTS')
# Verify Slack signature
import hmac
import hashlib
def verify_slack_signature(timestamp, body, signature, secret):
sig_basestring = f'v0:{timestamp}:{body}'
my_signature = 'v0=' + hmac.new(
secret.encode('utf-8'),
sig_basestring.encode('utf-8'),
hashlib.sha256
).hexdigest()
return hmac.compare_digest(my_signature, signature)
is_valid = verify_slack_signature(
request.headers['X-Slack-Request-Timestamp'],
request_body,
request.headers['X-Slack-Signature'],
webhook._Apikey # From Infisical
)
Infisical Setup:
Section: webhook.CUSTOM_INBOUND
username: webhook_user
password: secure_password_123
WebHooks Usage:
webhook = WebHooks()
webhook.Read('CUSTOM_INBOUND')
# Verify Basic Auth
import base64
def verify_basic_auth(auth_header, expected_user, expected_pass):
if not auth_header or not auth_header.startswith('Basic '):
return False
credentials = base64.b64decode(
auth_header[6:]
).decode('utf-8')
username, password = credentials.split(':', 1)
return (username == expected_user and
password == expected_pass)
is_valid = verify_basic_auth(
request.headers.get('Authorization'),
webhook._User, # From Infisical
webhook._Password # From Infisical
)
Infisical Setup:
Section: webhook.API_EVENTS
apikey: Bearer_token_xyz789abc456
WebHooks Usage:
webhook = WebHooks()
webhook.Read('API_EVENTS')
# Verify Bearer token
def verify_bearer_token(auth_header, expected_token):
if not auth_header or not auth_header.startswith('Bearer '):
return False
token = auth_header[7:]
return token == expected_token
is_valid = verify_bearer_token(
request.headers.get('Authorization'),
webhook._Apikey # From Infisical
)
Symptoms:
Causes:
Solution:
# 1. Check Infisical has the secret
python factory.core/ConfigIni.py infisical
# 2. Verify section name (must match exactly)
# Section: webhook.WEBHOOK_CODE (case-sensitive!)
# Key: password / username / apikey (lowercase!)
# 3. Test get_ini_value directly
python -c "
from factory.core import ObjData
obj = ObjData.ObjData()
value = obj.get_ini_value('webhook.STRIPE_PAYMENT', 'apikey', '')
print(f'Found: {len(value) if value else 0} characters')
"
Symptoms:
hook.Username / hook.Password / hook.ApiKey are emptyCauses:
Solution:
-- Check if database still has values
SELECT Webhookcode,
User,
CASE WHEN Password IS NULL THEN 'NULL' ELSE 'SET' END as Password,
CASE WHEN ApiKey IS NULL THEN 'NULL' ELSE 'SET' END as ApiKey
FROM def_webhook
WHERE Webhookcode = 'YOUR_WEBHOOK';
-- If database is empty and Infisical is empty, restore from backup
-- or reconfigure the webhook credentials
Expected Behavior:
No action needed - fallback is automatic.
# resource.test/pytests/factory.core/test_ObjHook.py
import pytest
from unittest.mock import Mock, patch
from factory.core.ObjHook import ObjHook
class TestWebhookInfisicalIntegration:
"""Test webhook Infisical integration."""
def test_credentials_from_infisical(self):
"""Test webhook reads credentials from Infisical."""
hook = ObjHook()
hook._Webhookcode = "TEST_WEBHOOK"
# Mock get_ini_value to simulate Infisical
with patch.object(hook, 'get_ini_value') as mock_get:
mock_get.return_value = "infisical_secret"
result = hook._get_webhook_credential("password", "db_password")
assert result == "infisical_secret"
mock_get.assert_called_once_with(
"webhook.TEST_WEBHOOK",
"password",
""
)
def test_fallback_to_database(self):
"""Test fallback to database when Infisical unavailable."""
hook = ObjHook()
hook._Webhookcode = "TEST_WEBHOOK"
# Mock get_ini_value to return empty (not in Infisical)
with patch.object(hook, 'get_ini_value') as mock_get:
mock_get.return_value = ""
result = hook._get_webhook_credential("password", "db_password")
assert result == "db_password"
def test_empty_db_and_infisical(self):
"""Test behavior when both sources are empty."""
hook = ObjHook()
hook._Webhookcode = "TEST_WEBHOOK"
with patch.object(hook, 'get_ini_value') as mock_get:
mock_get.return_value = ""
result = hook._get_webhook_credential("password", "")
assert result == ""
# resource.test/pytests/factory.web/test_WebHooks.py
import pytest
from unittest.mock import patch
from factory.web.WebHooks import WebHooks
class TestWebhookInfisicalIntegration:
"""Test inbound webhook Infisical integration."""
def test_get_webhook_credential_from_infisical(self):
"""Test reading inbound webhook token from Infisical."""
webhook = WebHooks(db_connection)
webhook._Webhookcode = "STRIPE_WEBHOOK"
with patch.object(webhook, 'get_ini_value') as mock_get_ini:
mock_get_ini.return_value = "infisical_token_secret"
result = webhook._get_webhook_credential(
"apikey",
"db_token"
)
assert result == "infisical_token_secret"
mock_get_ini.assert_called_once_with(
"webhook.STRIPE_WEBHOOK",
"apikey",
""
)
def test_stripe_webhook_signature_validation(self):
"""Test Stripe-style webhook validation."""
webhook = WebHooks(db_connection)
webhook._Webhookcode = "STRIPE_PAYMENT"
webhook._Authtype = "APIKEY"
with patch.object(webhook, 'get_ini_value') as mock_get_ini:
mock_get_ini.return_value = "whsec_stripe_signing_secret"
# Simulate Read loading the signing secret
webhook._Apikey = webhook._get_webhook_credential(
"apikey",
"db_secret"
)
assert webhook._Apikey == "whsec_stripe_signing_secret"
def test_failover_on_network_error(self):
"""Test failover to database on network error."""
webhook = WebHooks(db_connection)
webhook._Webhookcode = "TEST_WEBHOOK"
with patch.object(webhook, 'get_ini_value') as mock_get_ini:
mock_get_ini.side_effect = ConnectionError("Network unreachable")
result = webhook._get_webhook_credential(
"apikey",
"db_fallback_token"
)
assert result == "db_fallback_token"
# Run all webhook Infisical integration tests
pytest resource.test/pytests/factory.core/test_ObjHook.py::TestWebhookInfisicalIntegration -v
pytest resource.test/pytests/factory.web/test_WebHooks.py -v
# Run specific test
pytest resource.test/pytests/factory.core/test_ObjHook.py::TestWebhookInfisicalIntegration::test_credentials_from_infisical -v
# Run with integration marker (requires real Infisical)
pytest -m integration resource.test/pytests/factory.core/test_ObjHook.py -v
# resource.test/pytests/integration/test_webhook_real_infisical.py
import pytest
from factory.core.ObjHook import ObjHook
from factory.web.WebHooks import WebHooks
@pytest.mark.integration
class TestWebhookInfisicalReal:
"""Integration tests requiring real Infisical connection."""
def test_read_outbound_webhook_from_infisical(self):
"""Test reading real outbound webhook from Infisical."""
hook = ObjHook()
# Assumes TEST_WEBHOOK exists in def_webhook table
# and has credentials in Infisical
hook.read('TEST_WEBHOOK', context={})
# Credentials should be populated
assert hook.Username or hook.Password or hook.ApiKey
def test_read_inbound_webhook_from_infisical(self):
"""Test reading real inbound webhook from Infisical."""
webhook = WebHooks()
# Assumes TEST_WEBHOOK configured in Infisical
webhook.Read('TEST_WEBHOOK')
# Token should be loaded
assert hasattr(webhook, '_Apikey')
Total tests: 38 (9 outbound + 29 inbound)
Outbound webhook tests (factory.core/test_ObjHook.py):
Inbound webhook tests (factory.web/test_WebHooks.py):
# .gitignore should include:
config.yaml # If it contains secrets
*.env
*credentials*
# Infisical Environments:
- development → webhook.XXX.password = "dev_password"
- uat → webhook.XXX.password = "uat_password"
- production → webhook.XXX.password = "prod_password"
# Rotate webhook credentials in Infisical
# No code changes needed - update Infisical only
# 1. Generate new credential with provider
# 2. Update in Infisical
# 3. Test webhook
# 4. Revoke old credential
# View Infisical audit log
# Who accessed webhook.STRIPE_PAYMENT.apikey?
# When was it last accessed?
# Infisical access control:
- Developers: Read-only access to dev environment
- Ops: Read/write access to UAT environment
- Admins only: Access to production environment
# Success (Infisical)
"Using password from Infisical for webhook STRIPE_PAYMENT"
# Fallback (Database)
"Using password from database for webhook STRIPE_PAYMENT"
# Error (Connection issue)
"Could not read password from Infisical for STRIPE_PAYMENT: Connection timeout"
# Example alert rules
alerts:
- name: webhook_infisical_fallback_high
condition: fallback_rate > 10%
action: notify_ops_team
message: "High webhook Infisical fallback rate - check Infisical connectivity"
- name: webhook_using_database_credentials
condition: webhook.*.source == 'database'
action: notify_security_team
message: "Webhook still using database credentials - migration incomplete"
resource.notes/KEYCLOAK_ENHANCEMENTS_PHASE3.md - Additional auth enhancementsfactory.core/ConfigIni.py - Infisical integration implementationfactory.core/ObjData.py - get_ini_value() method documentationfactory.core/ObjHook.py - Webhook implementationFor questions or issues:
python factory.core/ConfigIni.py infisicalget_ini_value('webhook.XXX', 'password', '')Implementation Date: 2026-02-07
Author: Claude Code
Status: ✅ Production Ready