CVE-2021-37705 (Python, published 2021-08-13T21:15Z)
Description: value: OneFuzz is an open source self-hosted Fuzzing-As-A-Service platform. Starting with OneFuzz 2.12.0 or greater, an incomplete authorization check allows an authenticated user from any Azure Active Directory tenant to make authorized API calls to a vulnerable OneFuzz instance. To be vulnerable, a OneFuzz deployment must be both version 2.12.0 or greater and deployed with the non-default --multi_tenant_domain option. This can result in read/write access to private data such as software vulnerability and crash information, security testing tools and proprietary code and symbols. Via authorized API calls, this also enables tampering with existing data and unauthorized code execution on Azure compute resources. This issue is resolved starting in release 2.31.0, via the addition of application-level check of the bearer token's `issuer` against an administrator-configured allowlist. As a workaround users can restrict access to the tenant of a deployed OneFuzz instance < 2.31.0 by redeploying in the default configuration, which omits the `--multi_tenant_domain` option.
Severity: MEDIUM (CVSS v3 base score: 10)
Weakness: CWE-863 \u2014 Incorrect Authorization
The product performs an authorization check when an actor attempts to access a resource or perform an action, but it does not correctly perform the check. This allows attackers to bypass intended access restrictions.
Repository: https://github.com/microsoft/onefuzz
Files changed:
docs/webhook_events.md
src/api-service/__app__/info/__init__.py
src/api-service/__app__/negotiate/__init__.py
src/api-service/__app__/onefuzzlib/config.py
src/api-service/__app__/onefuzzlib/user_credentials.py
src/api-service/tests/test_auth_check.py
src/deployment/azuredeploy.json
src/deployment/deploy.py
src/deployment/set_admins.py
src/pytypes/extra/generate-docs.py
src/pytypes/onefuzztypes/models.py
src/pytypes/tests/test_instance_config_update.py
Commit message: Merge pull request from GHSA-q5vh-6whw-x745

* verify aad tenants, primarily needed in multi-tenant deployments

* add logging and fix trailing slash for issuer

* handle call_if* not supporting additional argument callbacks

* add logging

* include new datatype in webhook docs

* fix pytypes unit tests

Co-authored-by: Brian Caswell <bmc@shmoo.com>

Vulnerable code:
        "allow_pool_management": true
def main(req: func.HttpRequest) -> func.HttpResponse:
    return func.HttpResponse(
        connectionInfoJson,
        status_code=200,
        headers={"Content-type": "application/json"},
    )
            entry = cls()
from typing import Optional
    # This token has already been verified by the azure authentication layer
from uuid import uuid4
        self.assertTrue(can_modify_config_impl(InstanceConfig(), UserInfo()))
            can_modify_config_impl(InstanceConfig(), UserInfo(object_id=user1))
                InstanceConfig(admins=[user1]), UserInfo(object_id=user1)
            can_modify_config_impl(InstanceConfig(admins=[user1]), UserInfo())
                InstanceConfig(admins=[user1]), UserInfo(object_id=user2)
                InstanceConfig(allow_pool_management=True), UserInfo()
                InstanceConfig(allow_pool_management=True), UserInfo(object_id=user1)
                InstanceConfig(allow_pool_management=False, admins=[user1]),
                InstanceConfig(allow_pool_management=False, admins=[user1]), UserInfo()
                InstanceConfig(allow_pool_management=False, admins=[user1]),
from set_admins import update_admins
        admins: List[UUID]
    def set_admins(self) -> None:
        ("set_admins", Client.set_admins),
    parser.add_argument("admins", type=UUID, nargs="*")
    update_admins(table_service, args.resource_group, args.admins)
        EventInstanceConfigUpdated(config=InstanceConfig(admins=[UUID(int=0)])),
        no_admins = InstanceConfig(admins=None)
        with_admins = InstanceConfig(admins=[UUID(int=0)])
        with_admins_2 = InstanceConfig(admins=[UUID(int=1)])

Fixed code:
        "allow_pool_management": true,
        "allowed_aad_tenants": [
            "00000000-0000-0000-0000-000000000000"
        ]
                },
                "allowed_aad_tenants": {
                    "items": {
                        "format": "uuid",
                        "type": "string"
                    },
                    "title": "Allowed Aad Tenants",
                    "type": "array"
            "required": [
                "allowed_aad_tenants"
            ],
                },
                "allowed_aad_tenants": {
                    "items": {
                        "format": "uuid",
                        "type": "string"
                    },
                    "title": "Allowed Aad Tenants",
                    "type": "array"
            "required": [
                "allowed_aad_tenants"
            ],
from ..onefuzzlib.endpoint_authorization import call_if_user
def get(req: func.HttpRequest) -> func.HttpResponse:


def main(req: func.HttpRequest) -> func.HttpResponse:
    methods = {"GET": get}
    method = methods[req.method]
    result = call_if_user(req, method)

    return result
from ..onefuzzlib.endpoint_authorization import call_if_user

    # NOTE: this is a sub-method because the call_if* do not support callbacks with
    # additional arguments at this time.  Once call_if* supports additional arguments,
    # this should be made a generic function
    def post(req: func.HttpRequest) -> func.HttpResponse:
        return func.HttpResponse(
            connectionInfoJson,
            status_code=200,
            headers={"Content-type": "application/json"},
        )

    methods = {"POST": post}
    method = methods[req.method]

    result = call_if_user(req, method)

    return result
            entry = cls(allowed_aad_tenants=[])
import logging
from typing import List, Optional
from memoization import cached
from .config import InstanceConfig

@cached(ttl=60)
def get_allowed_tenants() -> List[str]:
    config = InstanceConfig.fetch()
    entries = [f"https://sts.windows.net/{x}/" for x in config.allowed_aad_tenants]
    return entries


    # The JWT token has already been verified by the azure authentication layer,
    # but we need to verify the tenant is as we expect.
    if "iss" not in token:
        return Error(
            code=ErrorCode.INVALID_REQUEST, errors=["missing issuer from token"]
        )

    tenants = get_allowed_tenants()
    if token["iss"] not in tenants:
        logging.error("issuer not from allowed tenant: %s - %s", token["iss"], tenants)
        return Error(code=ErrorCode.INVALID_REQUEST, errors=["unauthorized AAD issuer"])

from uuid import UUID, uuid4
        self.assertTrue(
            can_modify_config_impl(
                InstanceConfig(allowed_aad_tenants=[UUID(int=0)]), UserInfo()
            )
        )
            can_modify_config_impl(
                InstanceConfig(allowed_aad_tenants=[UUID(int=0)]),
                UserInfo(object_id=user1),
            )
                InstanceConfig(allowed_aad_tenants=[UUID(int=0)], admins=[user1]),
                UserInfo(object_id=user1),
            can_modify_config_impl(
                InstanceConfig(allowed_aad_tenants=[UUID(int=0)], admins=[user1]),
                UserInfo(),
            )
                InstanceConfig(allowed_aad_tenants=[UUID(int=0)], admins=[user1]),
                UserInfo(object_id=user2),
                InstanceConfig(
                    allowed_aad_tenants=[UUID(int=0)], allow_pool_management=True
                ),
                UserInfo(),
                InstanceConfig(
                    allowed_aad_tenants=[UUID(int=0)], allow_pool_management=True
                ),
                UserInfo(object_id=user1),
                InstanceConfig(
                    allowed_aad_tenants=[UUID(int=0)],
                    allow_pool_management=False,
                    admins=[user1],
                ),
                InstanceConfig(
                    allowed_aad_tenants=[UUID(int=0)],
                    allow_pool_management=False,
                    admins=[user1],
                ),
                UserInfo(),
                InstanceConfig(
                    allowed_aad_tenants=[UUID(int=0)],
                    allow_pool_management=False,
                    admins=[user1],
                ),
        },
        "tenant_id": {
            "type": "string",
            "value": "[subscription().tenantId]"
from set_admins import update_admins, update_allowed_aad_tenants
        admins: List[UUID],
        allowed_aad_tenants: List[UUID],
        self.allowed_aad_tenants = allowed_aad_tenants
    def set_instance_config(self) -> None:
        tenant = UUID(self.results["deploy"]["tenant_id"]["value"])

        tenants = self.allowed_aad_tenants
        if tenant not in tenants:
            tenants.append(tenant)
        update_allowed_aad_tenants(table_service, self.application_name, tenants)

        ("set_instance_config", Client.set_instance_config),
    parser.add_argument(
        "--allowed_aad_tenants",
        type=UUID,
        nargs="*",
        help="Set additional AAD tenants beyond the tenant the app is deployed in",
    )
        allowed_aad_tenants=args.allowed_aad_tenants or [],
def update_allowed_aad_tenants(
    table_service: TableService, resource_group: str, tenants: List[UUID]
) -> None:
    create_if_missing(table_service)
    as_str = [str(x) for x in tenants]
    table_service.insert_or_merge_entity(
        TABLE_NAME,
        {
            "PartitionKey": resource_group,
            "RowKey": resource_group,
            "allowed_aad_tenants": json.dumps(as_str),
        },
    )


    parser.add_argument("--admins", type=UUID, nargs="*")
    parser.add_argument("--allowed_aad_tenants", type=UUID, nargs="*")
    if args.admins:
        update_admins(table_service, args.resource_group, args.admins)
    if args.allowed_aad_tenants:
        update_allowed_aad_tenants(
            table_service, args.resource_group, args.allowed_aad_tenants
        )
        EventInstanceConfigUpdated(
            config=InstanceConfig(
                admins=[UUID(int=0)], allowed_aad_tenants=[UUID(int=0)]
            )
        ),
    allowed_aad_tenants: List[UUID]

    # At the moment, this only checks allowed_aad_tenants, however adding
    # support for 3rd party JWT validation is anticipated in a future release.
    @root_validator()
    def check_instance_config(cls, values: Any) -> Any:
        if "allowed_aad_tenants" not in values:
            raise ValueError("missing allowed_aad_tenants")

        if not len(values["allowed_aad_tenants"]):
            raise ValueError("allowed_aad_tenants must not be empty")
        return values

        no_admins = InstanceConfig(admins=None, allowed_aad_tenants=[UUID(int=0)])
        with_admins = InstanceConfig(
            admins=[UUID(int=0)], allowed_aad_tenants=[UUID(int=0)]
        )
        with_admins_2 = InstanceConfig(
            admins=[UUID(int=1)], allowed_aad_tenants=[UUID(int=0)]
        )