in src/tools/azure-cli-extension/cleanroom/cleanroom_common/azure_cleanroom_core/utilities/helpers.py [0:0]
def validate_config(spec: CleanRoomSpecification, logger: logging.Logger):
# TODO (HPrabh): Update the validate function to check the whole spec for anomalies.
issues = []
warnings = []
seen = set()
dupes = []
for application in spec.applications:
if application.datasources:
for datasource in application.datasources.keys():
index = next(
(i for i, x in enumerate(spec.datasources) if x.name == datasource),
None,
)
if index == None:
logger.error(
f"Datasource {datasource} not found in the cleanroom specification."
)
issues.append(
CleanroomSpecificationError(
ErrorCode.DatastoreNotFound,
f"Datasource {datasource} not found in the cleanroom specification.",
)
)
if application.datasinks:
for datasink in application.datasinks.keys():
index = next(
(i for i, x in enumerate(spec.datasinks) if x.name == datasink),
None,
)
if index == None:
logger.error(
f"Datasink {datasink} not found in the cleanroom specification."
)
issues.append(
CleanroomSpecificationError(
ErrorCode.DatasinkNotFound,
f"Datasink {datasink} not found in the cleanroom specification.",
)
)
if spec.network:
if spec.network.http:
if spec.network.http.inbound:
if not spec.network.http.inbound.policy.privacyPolicy:
warnings.append(
{
"code": "InboundAllowAll",
"message": "Inbound traffic is allowed. Configure a network policy to restrict traffic.",
}
)
else:
if len(seen) > 0:
warnings.append(
{
"code": "InboundTrafficNotAllowed",
"message": "Application ports are defined but no inbound traffic is disabled. "
+ "Please run `az cleanroom config network enable http` to enable inbound traffic.",
}
)
if spec.network.http.outbound:
if not spec.network.http.outbound.policy.privacyPolicy:
warnings.append(
{
"code": "OutboundAllowAll",
"message": "Outbound traffic is allowed. Configure a network policy to restrict traffic.",
}
)
if len(spec.applications) > 1:
warnings.append(
{
"code": "MultipleApplications",
"message": "Multiple applications are defined in the specification. "
+ "Please verify that the associated network policies handle expected ingress / egress.",
}
)
if len(dupes) > 0:
issues.append(
CleanroomSpecificationError(
ErrorCode.DuplicatePort,
f"Port {dupes} appear more than once in the application(s). "
+ "A port value can be used only once.",
)
)
return issues, warnings