Quick Azure/M365 Tenant Reconnaissance

msrecon.py

from argparse import ArgumentParser, BooleanOptionalAction
from typing import Iterable, TypedDict
import json
import sys
import urllib3
import xml.etree.ElementTree as ET

from requests import Session

# based on https://github.com/Gerenios/AADInternals/blob/b23a7845f6dc5ea8c57b10351421a4d00466cd90/KillChain.ps1#L8


def main() -> None:
    entrypoint = ArgumentParser()
    entrypoint.add_argument('--debug', action=BooleanOptionalAction, default=False)
    entrypoint.add_argument('domain', nargs=1)
    opts = entrypoint.parse_args()

    with Session() as session:
        if opts.debug:
            urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
            session.verify = False
            session.proxies.update(http='http://localhost:8080', https='http://localhost:8080')

        result = {}

        user_realm = get_user_realm(session, opts.domain[0])
        assert user_realm['NameSpaceType'] != 'Unknown'
        result['name'] = user_realm['FederationBrandName']

        openid_config = get_openid_config(session, opts.domain[0])
        result['id'] = openid_config['authorization_endpoint'].removeprefix('https://login.microsoftonline.com/').removesuffix('/oauth2/authorize')
        result['region'] = openid_config['tenant_region_scope']

        domains = get_tenant_domains(session, opts.domain[0])
        credential_types = {domain: get_credential_type(session, f'ADToAADSyncServiceAccount@{domain}') for domain in domains}
        result['stub'] = get_tenant_stub(domains)
        result['ssso'] = has_seamless_single_signon(credential_types.values())
        result['cloudsync'] = has_cloud_sync(credential_types.values())
        result['domains'] = []
        for domain in credential_types:
            user_realm = get_user_realm(session, domain)
            result['domains'].append(dict(
                domain=domain,
                type=user_realm['NameSpaceType'],  # if NameSpaceType equals managed, then the org has M365
            ))

        # TODO:
        # for each domain:
        # HasCloudMX, dig MX $domain
        # HasCloudSPF, dig TXT $domain | grep v=spf1
        # HasDMARC, dig TXT _dmarc.$domain

        # TODO:
        # GetMDIInstance, dig A $stub.atp.azure.com || dig A $stub-onmicrosoft-com.atp.azure.com

        print(json.dumps(result, sort_keys=False, indent=2))


class UserRealm(TypedDict):
  State: int
  UserState: int
  Login: str
  NameSpaceType: str
  DomainName: str
  FederationBrandName: str
  CloudInstanceName: str
  CloudInstanceIssuerUri: str


def get_user_realm(session: Session, domain: str) -> UserRealm:
    response = session.get(f'https://login.microsoftonline.com/getuserrealm.srf?login=test@{domain}')
    response.raise_for_status()
    data = response.json()
    return data


class OpenIdConfiguration(TypedDict):
    token_endpoint: str
    token_endpoint_auth_methods_supported: list[str]
    jwks_uri: str
    response_modes_supported: list[str]
    subject_types_supported: list[str]
    id_token_signing_alg_values_supported: list[str]
    response_types_supported: list[str]
    scopes_supported: list[str]
    issuer: str
    microsoft_multi_refresh_token: bool
    authorization_endpoint: str
    device_authorization_endpoint: str
    http_logout_supported: bool
    frontchannel_logout_supported: bool
    end_session_endpoint: str
    claims_supported: list[str]
    check_session_iframe: str
    userinfo_endpoint: str
    kerberos_endpoint: str
    tenant_region_scope: str
    cloud_instance_name: str
    cloud_graph_host_name: str
    msgraph_host: str
    rbac_url: str


def get_openid_config(session: Session, domain: str) -> OpenIdConfiguration:
    response = session.get(f'https://login.microsoftonline.com/{domain}/.well-known/openid-configuration')
    response.raise_for_status()
    data = response.json()
    return data


def get_tenant_domains(session: Session, domain: str) -> list[str]:
    body = f'''<?xml version="1.0" encoding="utf-8"?>
<soap:Envelope xmlns:exm="http://schemas.microsoft.com/exchange/services/2006/messages" xmlns:ext="http://schemas.microsoft.com/exchange/services/2006/types" xmlns:a="http://www.w3.org/2005/08/addressing" xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:xsd="http://www.w3.org/2001/XMLSchema">
  <soap:Header>
    <a:Action soap:mustUnderstand="1">http://schemas.microsoft.com/exchange/2010/Autodiscover/Autodiscover/GetFederationInformation</a:Action>
    <a:To soap:mustUnderstand="1">https://autodiscover-s.outlook.com/autodiscover/autodiscover.svc</a:To>
    <a:ReplyTo>
      <a:Address>http://www.w3.org/2005/08/addressing/anonymous</a:Address>
    </a:ReplyTo>
  </soap:Header>
  <soap:Body>
    <GetFederationInformationRequestMessage xmlns="http://schemas.microsoft.com/exchange/2010/Autodiscover">
      <Request>
        <Domain>{domain}</Domain>
      </Request>
    </GetFederationInformationRequestMessage>
  </soap:Body>
</soap:Envelope>'''
    response = session.post(
        'https://autodiscover-s.outlook.com/autodiscover/autodiscover.svc',
        headers={
            'Content-Type': 'text/xml;charset=utf-8',
            'SOAPAction': 'http://schemas.microsoft.com/exchange/2010/Autodiscover/Autodiscover/GetFederationInformation',
            'User-Agent': 'AutodiscoverClient',
        },
        data=body,
    )
    response.raise_for_status()
    root = ET.fromstring(response.text)
    namespaces = {
        's': 'http://schemas.xmlsoap.org/soap/envelope/',
        'h': 'http://schemas.microsoft.com/exchange/2010/Autodiscover',
    }
    domains = [d.text for d in root.iterfind('./s:Body/h:GetFederationInformationResponseMessage/h:Response/h:Domains/h:Domain', namespaces)]
    domains = [d for d in domains if d]
    return domains


def get_tenant_stub(domains: list[str]) -> str:
    suffix = '.onmicrosoft.com'
    results = [
        domain.removesuffix(suffix)
        for domain in domains
        if domain.endswith(suffix) and '.' not in domain.removesuffix('.onmicrosoft.com')
    ]
    if len(results) != 1:
        print(f'warning: could not determine tenant stub, found multiple candidates: {', '.join(results)}', file=sys.stderr)
    return results[0]


class CredentialTypeCredential(TypedDict):
    PrefCredential: int
    HasPassword: bool
    RemoteNgcParams: str
    FidoParams: str
    SasParams: str
    CertAuthParams: str
    GoogleParams: str
    FacebookParams: str


class CredentialTypeEstsPropertiesUserTenantBranding(TypedDict):
    Locale: int
    TileLogo: str
    TileDarkLogo: str
    Illustration: str
    BackgroundColor: str
    BoilerPlateText: str
    UserIdLabel: str
    KeepMeSignedInDisabled: bool
    UseTransparentLightBox: bool


class CredentialTypeEstsProperties(TypedDict):
    DesktopSsoEnabled: bool
    UserTenantBranding: list[CredentialTypeEstsPropertiesUserTenantBranding]
    DomainType: int


class CredentialType(TypedDict):
   Username: str
   Display: str
   IfExistsResult: int  # 0 if user exists else 1
   IsUnmanaged: bool
   ThrottleStatus: int
   Credentials: CredentialTypeCredential
   EstsProperties: CredentialTypeEstsProperties
   FlowToken: str
   IsSignupDisallowed: bool
   apiCanary: str


def get_credential_type(session: Session, email: str) -> CredentialType:
    response = session.get(
        'https://login.microsoftonline.com/common/GetCredentialType',
        json={
            'username': email,
            'isOtherIdpSupported': True,
            'checkPhones': True,
            'isRemoteNGCSupported': False,
            'isCookieBannerShown': False,
            'isFidoSupported': False,
            'originalRequest': '',
            'flowToken': '',
        }
    )
    response.raise_for_status()
    data = response.json()
    return data


def has_seamless_single_signon(credential_types: Iterable[CredentialType]) -> bool:
    return any(credential_type['EstsProperties']['DesktopSsoEnabled'] for credential_type in credential_types)


def has_cloud_sync(credential_types: Iterable[CredentialType]) -> bool:
    return any(credential_type['IfExistsResult'] == 0 for credential_type in credential_types)


if __name__ == '__main__':
    main()