19547151184

Committed 20 Nov 2025 06:20PM UTC coverage: 92.397% (+0.01%) from 92.384%

Build # 19547151184

Build Type

Pull #906

github

Committed by

web-flow

Commit Message

Merge a4fc548e8 into 4906a7603

Pull Request Pull Request #906: Feat/manage script testing

Run Details

15 of 16 new or added lines in 3 files covered. (93.75%)

28 existing lines in 2 files now uncovered.

1592 of 1723 relevant lines covered (92.4%)

0.92 hits per line

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

99.25

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

import json
import logging
import logging.config
import os
import sys
from enum import Enum
from functools import lru_cache
from pathlib import Path
from pydantic_settings import BaseSettings
from pydantic import ConfigDict
from typing import Any

import structlog


# Removed in later versions of python
def strtobool(val: str | bool) -> bool:
    """Convert a string representation of truth to a boolean (True or False).
    True values are 'y', 'yes', 't', 'true', 'on', and '1'; False
    values are 'n', 'no', 'f', 'false', 'off', and '0'. If val is
    already a boolean it is simply returned.  Raises ValueError if
    'val' is anything else.
    """
    if isinstance(val, bool):
        return val

    val = val.lower()
    if val in ("y", "yes", "t", "true", "on", "1"):
        return True
    elif val in ("n", "no", "f", "false", "off", "0"):
        return False
    else:
        raise ValueError(f"invalid truth value {val}")


# Use environment variable to determine logging format
# default to True
# strtobool will convert the results of the environment variable to a bool
use_json_logs: bool = strtobool(
    os.environ.get("LOG_WITH_JSON", True)
    if os.environ.get("LOG_WITH_JSON", True) != ""
    else True
)

time_stamp_format: str = os.environ.get("LOG_TIMESTAMP_FORMAT", "iso")

with open((Path(__file__).parent.parent / "logconf.json").resolve()) as user_file:
    file_contents: dict = json.loads(user_file.read())
    logging.config.dictConfig(file_contents["logger"])


def determin_log_level():
    match os.environ.get("LOG_LEVEL"):
        case "DEBUG":
            return logging.DEBUG
        case "INFO":
            return logging.INFO
        case "WARNING":
            return logging.WARNING
        case "ERROR":
            return logging.ERROR
        case _:
            return logging.DEBUG


logging.basicConfig(
    format="%(message)s",
    stream=sys.stdout,
    level=determin_log_level(),
)

shared_processors = [
    structlog.contextvars.merge_contextvars,
    structlog.stdlib.add_logger_name,
    structlog.stdlib.PositionalArgumentsFormatter(),
    structlog.stdlib.ExtraAdder(),
    structlog.processors.StackInfoRenderer(),
    structlog.stdlib.add_log_level,
    structlog.processors.TimeStamper(fmt=time_stamp_format),
]

renderer = (
    structlog.processors.JSONRenderer()
    if use_json_logs
    else structlog.dev.ConsoleRenderer()
)

# override uvicorn logging to use logstruct
formatter = structlog.stdlib.ProcessorFormatter(
    # These run ONLY on `logging` entries that do NOT originate within
    # structlog.
    foreign_pre_chain=shared_processors,
    # These run on ALL entries after the pre_chain is done.
    processors=[
        # Remove _record & _from_structlog.
        structlog.stdlib.ProcessorFormatter.remove_processors_meta,
        renderer,
    ],
)

handler = logging.StreamHandler()
handler.setFormatter(formatter)

for _log in ["uvicorn", "uvicorn.error"]:
    # Clear the log handlers for uvicorn loggers, and enable propagation
    # so the messages are caught by our root logger and formatted correctly
    # by structlog
    logging.getLogger(_log).handlers.clear()
    logging.getLogger(_log).addHandler(handler)
    logging.getLogger(_log).propagate = False

# This is already handled by our middleware
logging.getLogger("uvicorn.access").handlers.clear()
logging.getLogger("uvicorn.access").propagate = False

# Configure structlog
structlog.configure(
    processors=[structlog.stdlib.filter_by_level] + shared_processors + [renderer],
    context_class=dict,
    logger_factory=structlog.stdlib.LoggerFactory(),
    wrapper_class=structlog.make_filtering_bound_logger(
        logging.getLogger().getEffectiveLevel()
    ),
    cache_logger_on_first_use=True,
)

# Setup logger for config
logger: structlog.typing.FilteringBoundLogger = structlog.getLogger(__name__)


class EnvironmentEnum(str, Enum):
    PRODUCTION = "production"
    LOCAL = "local"


class GlobalConfig(BaseSettings):
    TITLE: str = os.environ.get(
        "CONTROLLER_APP_TITLE", "acapy-vc-authn-oidc Controller"
    )
    DESCRIPTION: str = os.environ.get(
        "CONTROLLER_APP_DESCRIPTION",
        "An oidc authentication solution for verification credentials",
    )

    ENVIRONMENT: EnvironmentEnum
    DEBUG: bool = False
    TESTING: bool = False
    TIMEZONE: str = "UTC"

    # the following defaults match up with default values in scripts/.env.example
    # these MUST be all set in non-local environments.
    DB_HOST: str = os.environ.get("DB_HOST", "localhost")
    DB_PORT: int | str = os.environ.get("DB_PORT", "27017")
    DB_NAME: str = os.environ.get("DB_NAME", "oidc-controller")
    DB_USER: str = os.environ.get("OIDC_CONTROLLER_DB_USER", "oidccontrolleruser")
    DB_PASS: str = os.environ.get("OIDC_CONTROLLER_DB_USER_PWD", "oidccontrollerpass")

    MONGODB_URL: str = (
        f"""mongodb://{DB_USER}:{DB_PASS}@{DB_HOST}:{DB_PORT}/{DB_NAME}?retryWrites=true&w=majority"""  # noqa: E501
    )

    CONTROLLER_URL: str | None = os.environ.get("CONTROLLER_URL")
    # Where to send users when trying to scan with their mobile camera (not a wallet)
    CONTROLLER_CAMERA_REDIRECT_URL: str | None = os.environ.get(
        "CONTROLLER_CAMERA_REDIRECT_URL"
    )
    # The number of seconds to wait for a presentation to be verified, Default: 10
    CONTROLLER_PRESENTATION_EXPIRE_TIME: int = os.environ.get(
        "CONTROLLER_PRESENTATION_EXPIRE_TIME", 10
    )

    # How long auth_sessions with matching the states in
    # CONTROLLER_SESSION_TIMEOUT_CONFIG_FILE are stored for in seconds
    CONTROLLER_PRESENTATION_CLEANUP_TIME: int = os.environ.get(
        "CONTROLLER_PRESENTATION_CLEANUP_TIME", 86400
    )

    # Presentation record cleanup configuration
    # How long to retain presentation records in hours (default: 24 hours)
    CONTROLLER_PRESENTATION_RECORD_RETENTION_HOURS: int = int(
        os.environ.get("CONTROLLER_PRESENTATION_RECORD_RETENTION_HOURS", 24)
    )

    # Resource limits for cleanup operations to prevent excessive processing
    # Maximum presentation records to process per cleanup cycle (default: 1000)
    CONTROLLER_CLEANUP_MAX_PRESENTATION_RECORDS: int = int(
        os.environ.get("CONTROLLER_CLEANUP_MAX_PRESENTATION_RECORDS", 1000)
    )
    # Maximum connections to process per cleanup cycle (default: 2000)
    CONTROLLER_CLEANUP_MAX_CONNECTIONS: int = int(
        os.environ.get("CONTROLLER_CLEANUP_MAX_CONNECTIONS", 2000)
    )

    CONTROLLER_SESSION_TIMEOUT_CONFIG_FILE: str | None = os.environ.get(
        "CONTROLLER_SESSION_TIMEOUT_CONFIG_FILE"
    )
    ACAPY_AGENT_URL: str | None = os.environ.get("ACAPY_AGENT_URL")
    if not ACAPY_AGENT_URL:
        logger.warning("ACAPY_AGENT_URL was not provided, agent will not be accessible")

    ACAPY_TENANCY: str = os.environ.get(
        "ACAPY_TENANCY", "single"
    )  # valid options are "multi" and "single"

    ACAPY_ADMIN_URL: str = os.environ.get("ACAPY_ADMIN_URL", "http://localhost:8031")

    ACAPY_PROOF_FORMAT: str = os.environ.get("ACAPY_PROOF_FORMAT", "indy")

    MT_ACAPY_WALLET_ID: str | None = os.environ.get("MT_ACAPY_WALLET_ID")
    MT_ACAPY_WALLET_KEY: str = os.environ.get("MT_ACAPY_WALLET_KEY", "random-key")

    ST_ACAPY_ADMIN_API_KEY_NAME: str | None = os.environ.get(
        "ST_ACAPY_ADMIN_API_KEY_NAME"
    )
    ST_ACAPY_ADMIN_API_KEY: str | None = os.environ.get("ST_ACAPY_ADMIN_API_KEY")
    DB_ECHO_LOG: bool = False

    DEFAULT_PAGE_SIZE: int | str = os.environ.get("DEFAULT_PAGE_SIZE", 10)

    # openssl rand -hex 32
    SIGNING_KEY_SIZE: int = os.environ.get("SIGNING_KEY_SIZE", 2048)
    # SIGNING_KEY_FILEPATH expects complete path including filename and extension.
    SIGNING_KEY_FILEPATH: str | None = os.environ.get("SIGNING_KEY_FILEPATH")
    SIGNING_KEY_ALGORITHM: str = os.environ.get("SIGNING_KEY_ALGORITHM", "RS256")
    SUBJECT_ID_HASH_SALT: str = os.environ.get("SUBJECT_ID_HASH_SALT", "test_hash_salt")

    # OIDC Client Settings
    OIDC_CLIENT_ID: str = os.environ.get("OIDC_CLIENT_ID", "keycloak")
    OIDC_CLIENT_NAME: str = os.environ.get("OIDC_CLIENT_NAME", "keycloak")
    OIDC_CLIENT_REDIRECT_URI: str = os.environ.get(
        "OIDC_CLIENT_REDIRECT_URI",
        "http://localhost:8880/auth/realms/vc-authn/broker/vc-authn/endpoint",
    )
    OIDC_CLIENT_SECRET: str = os.environ.get("OIDC_CLIENT_SECRET", "**********")

    # OIDC Controller Settings
    INVITATION_LABEL: str = os.environ.get("INVITATION_LABEL", "VC-AuthN")
    CONTROLLER_API_KEY: str = os.environ.get("CONTROLLER_API_KEY", "")
    USE_OOB_LOCAL_DID_SERVICE: bool = strtobool(
        os.environ.get("USE_OOB_LOCAL_DID_SERVICE", True)
    )
    USE_CONNECTION_BASED_VERIFICATION: bool = strtobool(
        os.environ.get("USE_CONNECTION_BASED_VERIFICATION", True)
    )
    WALLET_DEEP_LINK_PREFIX: str = os.environ.get(
        "WALLET_DEEP_LINK_PREFIX", "bcwallet://aries_proof-request"
    )
    SET_NON_REVOKED: bool = strtobool(os.environ.get("SET_NON_REVOKED", True))

    CONTROLLER_VARIABLE_SUBSTITUTION_OVERRIDE: str | None = os.environ.get(
        "CONTROLLER_VARIABLE_SUBSTITUTION_OVERRIDE"
    )
    CONTROLLER_TEMPLATE_DIR: str = os.environ.get(
        "CONTROLLER_TEMPLATE_DIR", "/app/controller-config/templates"
    )

    # Redis Configuration for multi-pod Socket.IO
    REDIS_HOST: str = os.environ.get("REDIS_HOST", "redis")
    REDIS_PORT: int = int(os.environ.get("REDIS_PORT", 6379))
    REDIS_PASSWORD: str | None = os.environ.get("REDIS_PASSWORD")
    REDIS_DB: int = int(os.environ.get("REDIS_DB", 0))
    USE_REDIS_ADAPTER: bool = strtobool(os.environ.get("USE_REDIS_ADAPTER", False))

    # Redis error handling and retry configuration
    REDIS_THREAD_MAX_RETRIES: int = int(os.environ.get("REDIS_THREAD_MAX_RETRIES", 5))
    REDIS_PUBSUB_MAX_FAILURES: int = int(
        os.environ.get("REDIS_PUBSUB_MAX_FAILURES", 10)
    )
    REDIS_RETRY_BASE_DELAY: int = int(os.environ.get("REDIS_RETRY_BASE_DELAY", 1))
    REDIS_RETRY_MAX_DELAY: int = int(os.environ.get("REDIS_RETRY_MAX_DELAY", 60))

    model_config = ConfigDict(case_sensitive=True)


class LocalConfig(GlobalConfig):
    """Local configurations."""

    DEBUG: bool = True
    DB_ECHO_LOG: bool = True
    ENVIRONMENT: EnvironmentEnum = EnvironmentEnum.LOCAL


class ProdConfig(GlobalConfig):
    """Production configurations."""

    DEBUG: bool = False
    ENVIRONMENT: EnvironmentEnum = EnvironmentEnum.PRODUCTION


class FactoryConfig:
    def __init__(self, environment: str | None):
        self.environment = environment

    def __call__(self) -> GlobalConfig:
        if self.environment == EnvironmentEnum.LOCAL.value:
            return LocalConfig()
        return ProdConfig()


@lru_cache()
def get_configuration() -> GlobalConfig:
    return FactoryConfig(os.environ.get("ENVIRONMENT"))()


settings = get_configuration()

# Add startup validation for ACAPY_PROOF_FORMAT
if settings.ACAPY_PROOF_FORMAT not in ["indy", "anoncreds"]:
    raise ValueError(
        f"ACAPY_PROOF_FORMAT must be 'indy' or 'anoncreds', got '{settings.ACAPY_PROOF_FORMAT}'"
    )

1	import json	1✔
2	import logging	1✔
3	import logging.config	1✔
4	import os	1✔
5	import sys	1✔
6	from enum import Enum	1✔
7	from functools import lru_cache	1✔
8	from pathlib import Path	1✔
9	from pydantic_settings import BaseSettings	1✔
10	from pydantic import ConfigDict	1✔
11	from typing import Any	1✔
12
13	import structlog	1✔
14
15
16	# Removed in later versions of python
17	def strtobool(val: str \| bool) -> bool:	1✔
18	"""Convert a string representation of truth to a boolean (True or False).
19	True values are 'y', 'yes', 't', 'true', 'on', and '1'; False
20	values are 'n', 'no', 'f', 'false', 'off', and '0'. If val is
21	already a boolean it is simply returned. Raises ValueError if
22	'val' is anything else.
23	"""
24	if isinstance(val, bool):	1✔
25	return val	1✔
26
27	val = val.lower()	1✔
28	if val in ("y", "yes", "t", "true", "on", "1"):	1✔
29	return True	1✔
30	elif val in ("n", "no", "f", "false", "off", "0"):	1✔
31	return False	1✔
32	else:
33	raise ValueError(f"invalid truth value {val}")	1✔
34
35
36	# Use environment variable to determine logging format
37	# default to True
38	# strtobool will convert the results of the environment variable to a bool
39	use_json_logs: bool = strtobool(	1✔
40	os.environ.get("LOG_WITH_JSON", True)
41	if os.environ.get("LOG_WITH_JSON", True) != ""
42	else True
43	)
44
45	time_stamp_format: str = os.environ.get("LOG_TIMESTAMP_FORMAT", "iso")	1✔
46
47	with open((Path(__file__).parent.parent / "logconf.json").resolve()) as user_file:	1✔
48	file_contents: dict = json.loads(user_file.read())	1✔
49	logging.config.dictConfig(file_contents["logger"])	1✔
50
51
52	def determin_log_level():	1✔
53	match os.environ.get("LOG_LEVEL"):	1✔
54	case "DEBUG":	1✔
55	return logging.DEBUG	1✔
56	case "INFO":	1✔
57	return logging.INFO	1✔
58	case "WARNING":	1✔
59	return logging.WARNING	1✔
60	case "ERROR":	1✔
61	return logging.ERROR	1✔
62	case _:	1✔
63	return logging.DEBUG	1✔
64
65
66	logging.basicConfig(	1✔
67	format="%(message)s",
68	stream=sys.stdout,
69	level=determin_log_level(),
70	)
71
72	shared_processors = [	1✔
73	structlog.contextvars.merge_contextvars,
74	structlog.stdlib.add_logger_name,
75	structlog.stdlib.PositionalArgumentsFormatter(),
76	structlog.stdlib.ExtraAdder(),
77	structlog.processors.StackInfoRenderer(),
78	structlog.stdlib.add_log_level,
79	structlog.processors.TimeStamper(fmt=time_stamp_format),
80	]
81
82	renderer = (	1✔
83	structlog.processors.JSONRenderer()
84	if use_json_logs
85	else structlog.dev.ConsoleRenderer()
86	)
87
88	# override uvicorn logging to use logstruct
89	formatter = structlog.stdlib.ProcessorFormatter(	1✔
90	# These run ONLY on `logging` entries that do NOT originate within
91	# structlog.
92	foreign_pre_chain=shared_processors,
93	# These run on ALL entries after the pre_chain is done.
94	processors=[
95	# Remove _record & _from_structlog.
96	structlog.stdlib.ProcessorFormatter.remove_processors_meta,
97	renderer,
98	],
99	)
100
101	handler = logging.StreamHandler()	1✔
102	handler.setFormatter(formatter)	1✔
103
104	for _log in ["uvicorn", "uvicorn.error"]:	1✔
105	# Clear the log handlers for uvicorn loggers, and enable propagation
106	# so the messages are caught by our root logger and formatted correctly
107	# by structlog
108	logging.getLogger(_log).handlers.clear()	1✔
109	logging.getLogger(_log).addHandler(handler)	1✔
110	logging.getLogger(_log).propagate = False	1✔
111
112	# This is already handled by our middleware
113	logging.getLogger("uvicorn.access").handlers.clear()	1✔
114	logging.getLogger("uvicorn.access").propagate = False	1✔
115
116	# Configure structlog
117	structlog.configure(	1✔
118	processors=[structlog.stdlib.filter_by_level] + shared_processors + [renderer],
119	context_class=dict,
120	logger_factory=structlog.stdlib.LoggerFactory(),
121	wrapper_class=structlog.make_filtering_bound_logger(
122	logging.getLogger().getEffectiveLevel()
123	),
124	cache_logger_on_first_use=True,
125	)
126
127	# Setup logger for config
128	logger: structlog.typing.FilteringBoundLogger = structlog.getLogger(__name__)	1✔
129
130
131	class EnvironmentEnum(str, Enum):	1✔
132	PRODUCTION = "production"	1✔
133	LOCAL = "local"	1✔
134
135
136	class GlobalConfig(BaseSettings):	1✔
137	TITLE: str = os.environ.get(	1✔
138	"CONTROLLER_APP_TITLE", "acapy-vc-authn-oidc Controller"
139	)
140	DESCRIPTION: str = os.environ.get(	1✔
141	"CONTROLLER_APP_DESCRIPTION",
142	"An oidc authentication solution for verification credentials",
143	)
144
145	ENVIRONMENT: EnvironmentEnum	1✔
146	DEBUG: bool = False	1✔
147	TESTING: bool = False	1✔
148	TIMEZONE: str = "UTC"	1✔
149
150	# the following defaults match up with default values in scripts/.env.example
151	# these MUST be all set in non-local environments.
152	DB_HOST: str = os.environ.get("DB_HOST", "localhost")	1✔
153	DB_PORT: int \| str = os.environ.get("DB_PORT", "27017")	1✔
154	DB_NAME: str = os.environ.get("DB_NAME", "oidc-controller")	1✔
155	DB_USER: str = os.environ.get("OIDC_CONTROLLER_DB_USER", "oidccontrolleruser")	1✔
156	DB_PASS: str = os.environ.get("OIDC_CONTROLLER_DB_USER_PWD", "oidccontrollerpass")	1✔
157
158	MONGODB_URL: str = (	1✔
159	f"""mongodb://{DB_USER}:{DB_PASS}@{DB_HOST}:{DB_PORT}/{DB_NAME}?retryWrites=true&w=majority""" # noqa: E501
160	)
161
162	CONTROLLER_URL: str \| None = os.environ.get("CONTROLLER_URL")	1✔
163	# Where to send users when trying to scan with their mobile camera (not a wallet)
164	CONTROLLER_CAMERA_REDIRECT_URL: str \| None = os.environ.get(	1✔
165	"CONTROLLER_CAMERA_REDIRECT_URL"
166	)
167	# The number of seconds to wait for a presentation to be verified, Default: 10
168	CONTROLLER_PRESENTATION_EXPIRE_TIME: int = os.environ.get(	1✔
169	"CONTROLLER_PRESENTATION_EXPIRE_TIME", 10
170	)
171
172	# How long auth_sessions with matching the states in
173	# CONTROLLER_SESSION_TIMEOUT_CONFIG_FILE are stored for in seconds
174	CONTROLLER_PRESENTATION_CLEANUP_TIME: int = os.environ.get(	1✔
175	"CONTROLLER_PRESENTATION_CLEANUP_TIME", 86400
176	)
177
178	# Presentation record cleanup configuration
179	# How long to retain presentation records in hours (default: 24 hours)
180	CONTROLLER_PRESENTATION_RECORD_RETENTION_HOURS: int = int(	1✔
181	os.environ.get("CONTROLLER_PRESENTATION_RECORD_RETENTION_HOURS", 24)
182	)
183
184	# Resource limits for cleanup operations to prevent excessive processing
185	# Maximum presentation records to process per cleanup cycle (default: 1000)
186	CONTROLLER_CLEANUP_MAX_PRESENTATION_RECORDS: int = int(	1✔
187	os.environ.get("CONTROLLER_CLEANUP_MAX_PRESENTATION_RECORDS", 1000)
188	)
189	# Maximum connections to process per cleanup cycle (default: 2000)
190	CONTROLLER_CLEANUP_MAX_CONNECTIONS: int = int(	1✔
191	os.environ.get("CONTROLLER_CLEANUP_MAX_CONNECTIONS", 2000)
192	)
193
194	CONTROLLER_SESSION_TIMEOUT_CONFIG_FILE: str \| None = os.environ.get(	1✔
195	"CONTROLLER_SESSION_TIMEOUT_CONFIG_FILE"
196	)
197	ACAPY_AGENT_URL: str \| None = os.environ.get("ACAPY_AGENT_URL")	1✔
198	if not ACAPY_AGENT_URL:	1✔
199	logger.warning("ACAPY_AGENT_URL was not provided, agent will not be accessible")	1✔
200
201	ACAPY_TENANCY: str = os.environ.get(	1✔
202	"ACAPY_TENANCY", "single"
203	) # valid options are "multi" and "single"
204
205	ACAPY_ADMIN_URL: str = os.environ.get("ACAPY_ADMIN_URL", "http://localhost:8031")	1✔
206
207	ACAPY_PROOF_FORMAT: str = os.environ.get("ACAPY_PROOF_FORMAT", "indy")	1✔
208
209	MT_ACAPY_WALLET_ID: str \| None = os.environ.get("MT_ACAPY_WALLET_ID")	1✔
210	MT_ACAPY_WALLET_KEY: str = os.environ.get("MT_ACAPY_WALLET_KEY", "random-key")	1✔
211
212	ST_ACAPY_ADMIN_API_KEY_NAME: str \| None = os.environ.get(	1✔
213	"ST_ACAPY_ADMIN_API_KEY_NAME"
214	)
215	ST_ACAPY_ADMIN_API_KEY: str \| None = os.environ.get("ST_ACAPY_ADMIN_API_KEY")	1✔
216	DB_ECHO_LOG: bool = False	1✔
217
218	DEFAULT_PAGE_SIZE: int \| str = os.environ.get("DEFAULT_PAGE_SIZE", 10)	1✔
219
220	# openssl rand -hex 32
221	SIGNING_KEY_SIZE: int = os.environ.get("SIGNING_KEY_SIZE", 2048)	1✔
222	# SIGNING_KEY_FILEPATH expects complete path including filename and extension.
223	SIGNING_KEY_FILEPATH: str \| None = os.environ.get("SIGNING_KEY_FILEPATH")	1✔
224	SIGNING_KEY_ALGORITHM: str = os.environ.get("SIGNING_KEY_ALGORITHM", "RS256")	1✔
225	SUBJECT_ID_HASH_SALT: str = os.environ.get("SUBJECT_ID_HASH_SALT", "test_hash_salt")	1✔
226
227	# OIDC Client Settings
228	OIDC_CLIENT_ID: str = os.environ.get("OIDC_CLIENT_ID", "keycloak")	1✔
229	OIDC_CLIENT_NAME: str = os.environ.get("OIDC_CLIENT_NAME", "keycloak")	1✔
230	OIDC_CLIENT_REDIRECT_URI: str = os.environ.get(	1✔
231	"OIDC_CLIENT_REDIRECT_URI",
232	"http://localhost:8880/auth/realms/vc-authn/broker/vc-authn/endpoint",
233	)
234	OIDC_CLIENT_SECRET: str = os.environ.get("OIDC_CLIENT_SECRET", "**********")	1✔
235
236	# OIDC Controller Settings
237	INVITATION_LABEL: str = os.environ.get("INVITATION_LABEL", "VC-AuthN")	1✔
238	CONTROLLER_API_KEY: str = os.environ.get("CONTROLLER_API_KEY", "")	1✔
239	USE_OOB_LOCAL_DID_SERVICE: bool = strtobool(	1✔
240	os.environ.get("USE_OOB_LOCAL_DID_SERVICE", True)
241	)
242	USE_CONNECTION_BASED_VERIFICATION: bool = strtobool(	1✔
243	os.environ.get("USE_CONNECTION_BASED_VERIFICATION", True)
244	)
245	WALLET_DEEP_LINK_PREFIX: str = os.environ.get(	1✔
246	"WALLET_DEEP_LINK_PREFIX", "bcwallet://aries_proof-request"
247	)
248	SET_NON_REVOKED: bool = strtobool(os.environ.get("SET_NON_REVOKED", True))	1✔
249
250	CONTROLLER_VARIABLE_SUBSTITUTION_OVERRIDE: str \| None = os.environ.get(	1✔
251	"CONTROLLER_VARIABLE_SUBSTITUTION_OVERRIDE"
252	)
253	CONTROLLER_TEMPLATE_DIR: str = os.environ.get(	1✔
254	"CONTROLLER_TEMPLATE_DIR", "/app/controller-config/templates"
255	)
256
257	# Redis Configuration for multi-pod Socket.IO
258	REDIS_HOST: str = os.environ.get("REDIS_HOST", "redis")	1✔
259	REDIS_PORT: int = int(os.environ.get("REDIS_PORT", 6379))	1✔
260	REDIS_PASSWORD: str \| None = os.environ.get("REDIS_PASSWORD")	1✔
261	REDIS_DB: int = int(os.environ.get("REDIS_DB", 0))	1✔
262	USE_REDIS_ADAPTER: bool = strtobool(os.environ.get("USE_REDIS_ADAPTER", False))	1✔
263
264	# Redis error handling and retry configuration
265	REDIS_THREAD_MAX_RETRIES: int = int(os.environ.get("REDIS_THREAD_MAX_RETRIES", 5))	1✔
266	REDIS_PUBSUB_MAX_FAILURES: int = int(	1✔
267	os.environ.get("REDIS_PUBSUB_MAX_FAILURES", 10)
268	)
269	REDIS_RETRY_BASE_DELAY: int = int(os.environ.get("REDIS_RETRY_BASE_DELAY", 1))	1✔
270	REDIS_RETRY_MAX_DELAY: int = int(os.environ.get("REDIS_RETRY_MAX_DELAY", 60))	1✔
271
272	model_config = ConfigDict(case_sensitive=True)	1✔
273
274
275	class LocalConfig(GlobalConfig):	1✔
276	"""Local configurations."""
277
278	DEBUG: bool = True	1✔
279	DB_ECHO_LOG: bool = True	1✔
280	ENVIRONMENT: EnvironmentEnum = EnvironmentEnum.LOCAL	1✔
281
282
283	class ProdConfig(GlobalConfig):	1✔
284	"""Production configurations."""
285
286	DEBUG: bool = False	1✔
287	ENVIRONMENT: EnvironmentEnum = EnvironmentEnum.PRODUCTION	1✔
288
289
290	class FactoryConfig:	1✔
291	def __init__(self, environment: str \| None):	1✔
292	self.environment = environment	1✔
293
294	def __call__(self) -> GlobalConfig:	1✔
295	if self.environment == EnvironmentEnum.LOCAL.value:	1✔
296	return LocalConfig()	1✔
297	return ProdConfig()	1✔
298
299
300	@lru_cache()	1✔
301	def get_configuration() -> GlobalConfig:	1✔
302	return FactoryConfig(os.environ.get("ENVIRONMENT"))()	1✔
303
304
305	settings = get_configuration()	1✔
306
307	# Add startup validation for ACAPY_PROOF_FORMAT
308	if settings.ACAPY_PROOF_FORMAT not in ["indy", "anoncreds"]:	1✔
NEW UNCOV 309	raise ValueError(	×
310	f"ACAPY_PROOF_FORMAT must be 'indy' or 'anoncreds', got '{settings.ACAPY_PROOF_FORMAT}'"
311	)

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

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