upload

napt.upload.manager

Upload orchestrator for NAPT Intune deployment.

Coordinates the full upload pipeline: loading recipe config, inferring the package path, authenticating, parsing the .intunewin file, building app metadata, and executing the Graph API upload flow.

Example

Upload a packaged app to Intune:

from pathlib import Path
from napt.upload import upload_package

result = upload_package(Path("recipes/Google/chrome.yaml"))
print(f"Created Intune app: {result.intune_app_id}")
print(f"App: {result.app_name} {result.version}")

upload_package

upload_package(recipe_path: Path, force: bool = False) -> UploadResult

Upload a packaged app to Microsoft Intune via the Graph API.

Loads the recipe config, infers the .intunewin package path, authenticates using the available Azure credential, parses encryption metadata from the package, and executes the full Graph API upload flow.

When intune.build_types is "both" (the default), two Intune app entries are created: an install entry (detection script only) and an update entry (detection + requirements scripts). Each entry is created, uploaded, and committed in sequence before moving to the next.

The package directory is inferred as packages/{app.id}/{version}/. Run 'napt package' before calling this function.

Authentication is automatic — no configuration required:

Developers: set AZURE_CLIENT_ID and AZURE_TENANT_ID, complete device code flow
CI/CD: set AZURE_CLIENT_ID, AZURE_CLIENT_SECRET, AZURE_TENANT_ID
Azure-hosted runners: assign a managed identity to the resource

Before any Graph call, the package's installer hash (from the build manifest) is verified against the pending release recorded in the app's deployment state, so what was recorded at discovery is byte-for-byte what ships. A hash mismatch aborts the upload. When no pending release is recorded, the upload proceeds with a warning — or fails when deployment.require_pending is enabled. On success, the deployment state records the published version, hash, and Intune app IDs, and a matching pending slot is cleared.

Re-running an upload is safe: existing NAPT-stamped apps matching this publish instance (recipe id, entry type, installer hash) are adopted — or their interrupted content upload resumed — instead of duplicated. Adoption keeps the app as it is; it does not re-send metadata or content. Pass force=True to update matched apps' metadata and upload a fresh content version (e.g., after changing PSADT commands or detection settings without a new installer release).

Parameters:

Name	Type	Description	Default
`recipe_path`	`Path`	Path to the recipe YAML file.	required
`force`	`bool`	When True, matched stamped apps are re-uploaded (metadata and content) instead of adopted as-is. Never creates duplicates.	`False`

Returns:

Type	Description
`UploadResult`	Upload result including the Intune app ID(s), app name, version, and package path. intune_app_id is None when build_types is "update_only"; intune_update_app_id is None when build_types is "app_only".

Raises:

Type	Description
`ConfigError`	If the package directory is not found, or detection/ requirements scripts are absent from the package directory. Run 'napt package' to create or recreate the package.
`AuthError`	If all Azure credential methods fail.
`NetworkError`	If Graph API or Azure Blob Storage calls fail.
`PackagingError`	If the .intunewin file is malformed, the package's installer hash does not match the pending release in deployment state, or no pending release is recorded while deployment.require_pending is enabled.
`StateError`	On a corrupted deployment state file.

Example

Upload and print the resulting Intune app IDs:

from pathlib import Path
from napt.upload import upload_package

result = upload_package(Path("recipes/Google/chrome.yaml"))
print(f"Install app ID: {result.intune_app_id}")
if result.intune_update_app_id:
    print(f"Update app ID: {result.intune_update_app_id}")

Source code in napt/upload/manager.py

def upload_package(recipe_path: Path, force: bool = False) -> UploadResult:
    """Upload a packaged app to Microsoft Intune via the Graph API.

    Loads the recipe config, infers the .intunewin package path, authenticates
    using the available Azure credential, parses encryption metadata from the
    package, and executes the full Graph API upload flow.

    When intune.build_types is "both" (the default), two Intune app entries are
    created: an install entry (detection script only) and an update entry
    (detection + requirements scripts). Each entry is created, uploaded, and
    committed in sequence before moving to the next.

    The package directory is inferred as packages/{app.id}/{version}/.
    Run 'napt package' before calling this function.

    Authentication is automatic — no configuration required:

    - Developers: set AZURE_CLIENT_ID and AZURE_TENANT_ID, complete device code flow
    - CI/CD: set AZURE_CLIENT_ID, AZURE_CLIENT_SECRET, AZURE_TENANT_ID
    - Azure-hosted runners: assign a managed identity to the resource

    Before any Graph call, the package's installer hash (from the build
    manifest) is verified against the pending release recorded in the app's
    deployment state, so what was recorded at discovery is byte-for-byte
    what ships. A hash mismatch aborts the upload. When no pending release
    is recorded, the upload proceeds with a warning — or fails when
    deployment.require_pending is enabled. On success, the deployment
    state records the published version, hash, and Intune app IDs, and a
    matching pending slot is cleared.

    Re-running an upload is safe: existing NAPT-stamped apps matching this
    publish instance (recipe id, entry type, installer hash) are adopted —
    or their interrupted content upload resumed — instead of duplicated.
    Adoption keeps the app as it is; it does not re-send metadata or
    content. Pass force=True to update matched apps' metadata and upload a
    fresh content version (e.g., after changing PSADT commands or detection
    settings without a new installer release).

    Args:
        recipe_path: Path to the recipe YAML file.
        force: When True, matched stamped apps are re-uploaded (metadata
            and content) instead of adopted as-is. Never creates
            duplicates.

    Returns:
        Upload result including the Intune app ID(s), app name, version, and
            package path. intune_app_id is None when build_types is "update_only";
            intune_update_app_id is None when build_types is "app_only".

    Raises:
        ConfigError: If the package directory is not found, or detection/
            requirements scripts are absent from the package directory.
            Run 'napt package' to create or recreate the package.
        AuthError: If all Azure credential methods fail.
        NetworkError: If Graph API or Azure Blob Storage calls fail.
        PackagingError: If the .intunewin file is malformed, the
            package's installer hash does not match the pending release
            in deployment state, or no pending release is recorded while
            deployment.require_pending is enabled.
        StateError: On a corrupted deployment state file.

    Example:
        Upload and print the resulting Intune app IDs:
            ```python
            from pathlib import Path
            from napt.upload import upload_package

            result = upload_package(Path("recipes/Google/chrome.yaml"))
            print(f"Install app ID: {result.intune_app_id}")
            if result.intune_update_app_id:
                print(f"Update app ID: {result.intune_update_app_id}")
            ```

    """
    logger = get_global_logger()

    config = load_effective_config(recipe_path)
    app_id: str = config["id"]
    app_name: str = config["name"]
    build_types: str = config["intune"]["build_types"]

    logger.verbose("UPLOAD", f"Starting upload for '{app_name}' ({app_id})")
    logger.verbose("UPLOAD", f"build_types: {build_types}")

    # Resolve the app icon once; it is shared by the install and update entries
    large_icon = _resolve_large_icon(config)

    # Step 1: Locate the package directory
    total_steps = 9 if build_types == "both" else 6
    logger.step(1, total_steps, "Locating .intunewin package...")
    package_path, version = _infer_package_dir(app_id)
    logger.verbose("UPLOAD", f"Package: {package_path}")
    logger.verbose("UPLOAD", f"Version: {version}")

    manifest = _read_build_manifest(package_path.parent)
    installer_sha256: str = manifest["installer_sha256"]

    # Verify provenance against deployment state before any Graph call:
    # what was recorded at discovery must be byte-for-byte what ships.
    state_path = deployment_state_path(
        Path(config["directories"]["state"]) / "deployment", app_id
    )
    state = load_deployment_state(state_path)
    pending = state.get("pending")
    if pending:
        if pending.get("sha256") != installer_sha256:
            raise PackagingError(
                f"Installer hash mismatch for '{app_id}': the package was "
                f"built from a different binary than the pending release "
                f"recorded in {state_path}.\n"
                f"  pending:  {pending.get('version')} "
                f"(sha256 {pending.get('sha256')})\n"
                f"  package:  {version} (sha256 {installer_sha256})\n"
                "Re-run 'napt discover', 'napt build', and 'napt package' "
                "so the package matches the recorded release."
            )
        logger.info(
            "UPLOAD", f"Package matches pending release (sha256 {installer_sha256})"
        )
    elif config["deployment"]["require_pending"]:
        raise PackagingError(
            f"No pending release recorded for '{app_id}' and "
            "deployment.require_pending is enabled.\n"
            "Run 'napt discover' to record the release, or add a pending "
            f"entry (version, sha256, url) to {state_path}."
        )
    else:
        logger.warning(
            "UPLOAD",
            f"No pending release recorded for '{app_id}'; uploading "
            "without provenance verification.",
        )

    # Step 2: Authenticate
    logger.step(2, total_steps, "Authenticating with Azure...")
    access_token = get_access_token()

    # Step 3: Parse .intunewin metadata
    logger.step(3, total_steps, "Parsing package metadata...")
    intunewin_metadata = parse_intunewin(package_path)

    # Reconcile-before-act: list existing apps once so stamped apps from a
    # previous (possibly crashed) run are adopted instead of duplicated.
    existing_apps = list_mobile_apps(access_token)
    logger.verbose("UPLOAD", f"Tenant has {len(existing_apps)} mobile apps")

    intune_app_id: str | None = None
    intune_update_app_id: str | None = None

    if build_types in ("app_only", "both"):
        # Install entry: steps 4-6
        install_metadata = _build_app_metadata(
            config, recipe_path, version, package_path, "app_only", manifest, large_icon
        )
        intune_app_id = _upload_single_app(
            access_token,
            install_metadata,
            package_path,
            intunewin_metadata,
            existing_apps,
            recipe_id=app_id,
            entry=ENTRY_INSTALL,
            installer_sha256=installer_sha256,
            step_create=4,
            step_upload=5,
            step_commit=6,
            total_steps=total_steps,
            force=force,
        )

    if build_types in ("update_only", "both"):
        # Update entry: steps 4-6 (single) or 7-9 (both)
        step_offset = 6 if build_types == "both" else 3
        update_metadata = _build_app_metadata(
            config,
            recipe_path,
            version,
            package_path,
            "update_only",
            manifest,
            large_icon,
        )
        intune_update_app_id = _upload_single_app(
            access_token,
            update_metadata,
            package_path,
            intunewin_metadata,
            existing_apps,
            recipe_id=app_id,
            entry=ENTRY_UPDATE,
            installer_sha256=installer_sha256,
            step_create=step_offset + 1,
            step_upload=step_offset + 2,
            step_commit=step_offset + 3,
            total_steps=total_steps,
            force=force,
        )

    # Record the publication in deployment state: published version,
    # hash, and Intune app IDs; a matching pending slot is cleared.
    record_published(
        state,
        version=version,
        sha256=installer_sha256,
        intune_app_id=intune_app_id,
        intune_update_app_id=intune_update_app_id,
    )
    state["name"] = config["name"]
    save_deployment_state(state, state_path)
    logger.info("STATE", f"Recorded published release {version} in {state_path}")

    logger.verbose("UPLOAD", "Upload complete")

    return UploadResult(
        app_id=app_id,
        app_name=app_name,
        version=version,
        intune_app_id=intune_app_id,
        intune_update_app_id=intune_update_app_id,
        package_path=package_path,
        status="success",
    )

napt.upload.intunewin

Parses .intunewin package files for NAPT upload operations.

A .intunewin file is a ZIP archive created by IntuneWinAppUtil with the following structure:

IntuneWinPackage/
  Contents/
    IntunePackage.intunewin   <- encrypted payload
  Metadata/
    Detection.xml             <- encryption metadata

This module extracts the encryption metadata from Detection.xml and provides utilities for extracting the encrypted payload for upload to Azure Blob Storage.

Example

Parse metadata and extract payload:

from pathlib import Path
from napt.upload.intunewin import parse_intunewin, extract_encrypted_payload

metadata = parse_intunewin(
    Path("packages/napt-chrome/Invoke-AppDeployToolkit.intunewin")
)
print(f"Encrypted file: {metadata.encrypted_file_name}")
print(f"Encryption key: {metadata.encryption_key}")

IntunewinMetadata `dataclass`

Encryption metadata extracted from a .intunewin package.

All fields are sourced from Detection.xml inside the .intunewin ZIP archive. This metadata is required by the Graph API file commit endpoint.

Attributes:

Name	Type	Description
`encrypted_file_name`	`str`	Filename of the encrypted payload inside the Contents/ directory (always "IntunePackage.intunewin").
`unencrypted_content_size`	`int`	Original size in bytes before encryption.
`file_digest`	`str`	Base64-encoded SHA-256 hash of the encrypted payload.
`file_digest_algorithm`	`str`	Hash algorithm used (always "SHA256").
`encryption_key`	`str`	Base64-encoded AES-256 encryption key.
`mac_key`	`str`	Base64-encoded HMAC key for MAC verification.
`init_vector`	`str`	Base64-encoded AES initialization vector.
`mac`	`str`	Base64-encoded MAC value for integrity verification.
`profile_identifier`	`str`	Encryption profile version (always "ProfileVersion1").
`encrypted_file_size`	`int`	Byte size of the encrypted payload file.

Source code in napt/upload/intunewin.py

@dataclass(frozen=True)
class IntunewinMetadata:
    """Encryption metadata extracted from a .intunewin package.

    All fields are sourced from Detection.xml inside the .intunewin ZIP archive.
    This metadata is required by the Graph API file commit endpoint.

    Attributes:
        encrypted_file_name: Filename of the encrypted payload inside the
            Contents/ directory (always "IntunePackage.intunewin").
        unencrypted_content_size: Original size in bytes before encryption.
        file_digest: Base64-encoded SHA-256 hash of the encrypted payload.
        file_digest_algorithm: Hash algorithm used (always "SHA256").
        encryption_key: Base64-encoded AES-256 encryption key.
        mac_key: Base64-encoded HMAC key for MAC verification.
        init_vector: Base64-encoded AES initialization vector.
        mac: Base64-encoded MAC value for integrity verification.
        profile_identifier: Encryption profile version (always "ProfileVersion1").
        encrypted_file_size: Byte size of the encrypted payload file.
    """

    encrypted_file_name: str
    unencrypted_content_size: int
    file_digest: str
    file_digest_algorithm: str
    encryption_key: str
    mac_key: str
    init_vector: str
    mac: str
    profile_identifier: str
    encrypted_file_size: int

parse_intunewin

parse_intunewin(intunewin_path: Path) -> IntunewinMetadata

Parse a .intunewin package and extract encryption metadata.

Reads IntuneWinPackage/Metadata/Detection.xml from inside the .intunewin ZIP and returns all encryption fields required for the Graph API upload flow.

Parameters:

Name	Type	Description	Default
`intunewin_path`	`Path`	Path to the .intunewin file to parse.	required

Returns:

Type	Description
`IntunewinMetadata`	Parsed encryption metadata from Detection.xml.

Raises:

Type	Description
`PackagingError`	If the file is not a valid ZIP, Detection.xml is missing, or required XML fields are absent or malformed.

Example

Parse an existing package:

from pathlib import Path
from napt.upload.intunewin import parse_intunewin

metadata = parse_intunewin(
    Path("packages/napt-chrome/Invoke-AppDeployToolkit.intunewin")
)
print(metadata.encryption_key)

Source code in napt/upload/intunewin.py

def parse_intunewin(intunewin_path: Path) -> IntunewinMetadata:
    """Parse a .intunewin package and extract encryption metadata.

    Reads IntuneWinPackage/Metadata/Detection.xml from inside the .intunewin
    ZIP and returns all encryption fields required for the Graph API upload flow.

    Args:
        intunewin_path: Path to the .intunewin file to parse.

    Returns:
        Parsed encryption metadata from Detection.xml.

    Raises:
        PackagingError: If the file is not a valid ZIP, Detection.xml is missing,
            or required XML fields are absent or malformed.

    Example:
        Parse an existing package:
            ```python
            from pathlib import Path
            from napt.upload.intunewin import parse_intunewin

            metadata = parse_intunewin(
                Path("packages/napt-chrome/Invoke-AppDeployToolkit.intunewin")
            )
            print(metadata.encryption_key)
            ```

    """
    try:
        zf = zipfile.ZipFile(intunewin_path, "r")
    except zipfile.BadZipFile as err:
        raise PackagingError(
            f"{intunewin_path} is not a valid .intunewin file (invalid ZIP archive)"
        ) from err
    except OSError as err:
        raise PackagingError(f"Failed to open {intunewin_path}: {err}") from err

    with zf:
        # Read Detection.xml
        try:
            xml_bytes = zf.read(DETECTION_XML_PATH)
        except KeyError as err:
            raise PackagingError(
                f"{intunewin_path} is missing {DETECTION_XML_PATH}. "
                "The file may be corrupt or was not created by IntuneWinAppUtil."
            ) from err

        # Get encrypted payload file size
        try:
            payload_info = zf.getinfo(ENCRYPTED_PAYLOAD_PATH)
            encrypted_file_size = payload_info.file_size
        except KeyError as err:
            raise PackagingError(
                f"{intunewin_path} is missing {ENCRYPTED_PAYLOAD_PATH}. "
                "The file may be corrupt or was not created by IntuneWinAppUtil."
            ) from err

    # Parse XML
    try:
        root = ET.fromstring(xml_bytes)
    except ET.ParseError as err:
        raise PackagingError(f"Detection.xml contains invalid XML: {err}") from err

    # Extract namespace from root tag (e.g., '{http://schemas.microsoft.com/...}')
    ns = ""
    if root.tag.startswith("{"):
        ns = root.tag[: root.tag.index("}") + 1]

    # Read top-level fields
    encrypted_file_name = _require_text(root, "FileName", ns, "FileName")
    unencrypted_size_str = _require_text(
        root, "UnencryptedContentSize", ns, "UnencryptedContentSize"
    )
    try:
        unencrypted_content_size = int(unencrypted_size_str)
    except ValueError as err:
        raise PackagingError(
            f"Detection.xml UnencryptedContentSize is not an integer: "
            f"'{unencrypted_size_str}'"
        ) from err

    # Read EncryptionInfo subsection
    enc_info = root.find(f"{ns}EncryptionInfo")
    if enc_info is None:
        raise PackagingError(
            "Detection.xml is missing required section 'EncryptionInfo'. "
            "The .intunewin file may be corrupt."
        )

    encryption_key = _require_text(
        enc_info, "EncryptionKey", ns, "EncryptionInfo/EncryptionKey"
    )
    mac_key = _require_text(enc_info, "MacKey", ns, "EncryptionInfo/MacKey")
    init_vector = _require_text(
        enc_info, "InitializationVector", ns, "EncryptionInfo/InitializationVector"
    )
    mac = _require_text(enc_info, "Mac", ns, "EncryptionInfo/Mac")
    profile_identifier = _require_text(
        enc_info, "ProfileIdentifier", ns, "EncryptionInfo/ProfileIdentifier"
    )
    file_digest = _require_text(enc_info, "FileDigest", ns, "EncryptionInfo/FileDigest")
    file_digest_algorithm = _require_text(
        enc_info, "FileDigestAlgorithm", ns, "EncryptionInfo/FileDigestAlgorithm"
    )

    return IntunewinMetadata(
        encrypted_file_name=encrypted_file_name,
        unencrypted_content_size=unencrypted_content_size,
        file_digest=file_digest,
        file_digest_algorithm=file_digest_algorithm,
        encryption_key=encryption_key,
        mac_key=mac_key,
        init_vector=init_vector,
        mac=mac,
        profile_identifier=profile_identifier,
        encrypted_file_size=encrypted_file_size,
    )

extract_encrypted_payload

extract_encrypted_payload(intunewin_path: Path, dest_dir: Path) -> Path

Extract the encrypted payload from a .intunewin package.

Extracts IntuneWinPackage/Contents/IntunePackage.intunewin to the destination directory for upload to Azure Blob Storage.

Parameters:

Name	Type	Description	Default
`intunewin_path`	`Path`	Path to the .intunewin file.	required
`dest_dir`	`Path`	Directory to extract the payload into.	required

Returns:

Type	Description
`Path`	Path to the extracted encrypted payload file.

Raises:

Type	Description
`PackagingError`	If the file is not a valid ZIP or the payload is missing.

Source code in napt/upload/intunewin.py

def extract_encrypted_payload(intunewin_path: Path, dest_dir: Path) -> Path:
    """Extract the encrypted payload from a .intunewin package.

    Extracts IntuneWinPackage/Contents/IntunePackage.intunewin to the
    destination directory for upload to Azure Blob Storage.

    Args:
        intunewin_path: Path to the .intunewin file.
        dest_dir: Directory to extract the payload into.

    Returns:
        Path to the extracted encrypted payload file.

    Raises:
        PackagingError: If the file is not a valid ZIP or the payload is missing.

    """
    try:
        zf = zipfile.ZipFile(intunewin_path, "r")
    except zipfile.BadZipFile as err:
        raise PackagingError(
            f"{intunewin_path} is not a valid .intunewin file (invalid ZIP archive)"
        ) from err
    except OSError as err:
        raise PackagingError(f"Failed to open {intunewin_path}: {err}") from err

    with zf:
        try:
            zf.extract(ENCRYPTED_PAYLOAD_PATH, dest_dir)
        except KeyError as err:
            raise PackagingError(
                f"{intunewin_path} is missing {ENCRYPTED_PAYLOAD_PATH}. "
                "The file may be corrupt or was not created by IntuneWinAppUtil."
            ) from err

    # zipfile.extract preserves the full path structure inside dest_dir
    return dest_dir / ENCRYPTED_PAYLOAD_PATH

napt.upload.auth

Azure credential acquisition for NAPT Intune upload.

Requires a NAPT app registration in Microsoft Entra ID with the DeviceManagementApps.ReadWrite.All Microsoft Graph API permission. See the authentication documentation for setup instructions.

Authentication is selected automatically based on environment variables:

Authentication order

EnvironmentCredential -- service principal via environment variables. Set AZURE_CLIENT_ID, AZURE_CLIENT_SECRET, and AZURE_TENANT_ID. Recommended for CI/CD pipelines (GitHub Actions, Azure DevOps, etc.).
ManagedIdentityCredential -- Azure managed identity. Works automatically on Azure VMs, Azure Container Instances, and Azure-hosted pipeline agents with a managed identity assigned. No credentials to manage.
DeviceCodeCredential -- interactive device code flow (TTY only). Requires AZURE_CLIENT_ID and AZURE_TENANT_ID to be set (no secret needed). Prints a URL and code; the user completes authentication in any browser. Skipped in CI/CD and when output is redirected.

If all available methods fail, an AuthError is raised with guidance on which environment variables to set.

Example

Acquiring a token for Graph API:

from napt.upload.auth import get_access_token

token = get_access_token()
headers = {"Authorization": f"Bearer {token}"}

get_credential

get_credential() -> ChainedTokenCredential

Build the Phase 1 credential chain for non-interactive authentication.

Returns a credential that tries service principal auth (via environment variables) first, then managed identity for Azure-hosted workloads. Both use the .default scope, suitable for application permissions.

For interactive device code auth, use get_access_token() directly, which handles Phase 2 automatically when Phase 1 fails.

Returns:

Type	Description
`ChainedTokenCredential`	A ChainedTokenCredential for non-interactive authentication.

Source code in napt/upload/auth.py

def get_credential() -> ChainedTokenCredential:
    """Build the Phase 1 credential chain for non-interactive authentication.

    Returns a credential that tries service principal auth (via environment
    variables) first, then managed identity for Azure-hosted workloads.
    Both use the `.default` scope, suitable for application permissions.

    For interactive device code auth, use `get_access_token()` directly,
    which handles Phase 2 automatically when Phase 1 fails.

    Returns:
        A ChainedTokenCredential for non-interactive authentication.

    """
    return ChainedTokenCredential(
        EnvironmentCredential(),
        ManagedIdentityCredential(),
    )

get_access_token

get_access_token() -> str

Acquire a Microsoft Graph API access token.

Tries credential methods in order until one succeeds:

Phase 1 (always tried): EnvironmentCredential (service principal via AZURE_CLIENT_ID, AZURE_CLIENT_SECRET, AZURE_TENANT_ID) then ManagedIdentityCredential. Both use the .default scope (application permissions).

Phase 2 (only if Phase 1 fails and stdout is a TTY): DeviceCodeCredential using AZURE_CLIENT_ID and AZURE_TENANT_ID. Uses the explicit DeviceManagementApps.ReadWrite.All scope, which triggers a consent prompt on first run.

Returns:

Type	Description
`str`	Bearer token string for use in Authorization headers.

Raises:

Type	Description
`AuthError`	If all credential types fail or are unavailable, with guidance on which environment variables to set.

Example

Get a token and use it in a request:

from napt.upload.auth import get_access_token

token = get_access_token()
headers = {"Authorization": f"Bearer {token}"}

Source code in napt/upload/auth.py

def get_access_token() -> str:
    """Acquire a Microsoft Graph API access token.

    Tries credential methods in order until one succeeds:

    Phase 1 (always tried):
        EnvironmentCredential (service principal via AZURE_CLIENT_ID,
        AZURE_CLIENT_SECRET, AZURE_TENANT_ID) then ManagedIdentityCredential.
        Both use the `.default` scope (application permissions).

    Phase 2 (only if Phase 1 fails and stdout is a TTY):
        DeviceCodeCredential using AZURE_CLIENT_ID and AZURE_TENANT_ID.
        Uses the explicit DeviceManagementApps.ReadWrite.All scope, which
        triggers a consent prompt on first run.

    Returns:
        Bearer token string for use in Authorization headers.

    Raises:
        AuthError: If all credential types fail or are unavailable,
            with guidance on which environment variables to set.

    Example:
        Get a token and use it in a request:
            ```python
            from napt.upload.auth import get_access_token

            token = get_access_token()
            headers = {"Authorization": f"Bearer {token}"}
            ```

    """
    # Phase 1: service principal or managed identity
    try:
        return get_credential().get_token(*GRAPH_SCOPES).token
    except ClientAuthenticationError:
        pass

    # Phase 2: interactive device code (TTY only)
    if sys.stdout.isatty():
        client_id = os.environ.get("AZURE_CLIENT_ID", "")
        tenant_id = os.environ.get("AZURE_TENANT_ID", "")
        if client_id and tenant_id:
            try:
                return (
                    DeviceCodeCredential(
                        client_id=client_id,
                        tenant_id=tenant_id,
                    )
                    .get_token(*_DEVICE_CODE_SCOPES)
                    .token
                )
            except ClientAuthenticationError as err:
                raise AuthError(
                    f"{_AUTH_FAILURE_HINT_DEVICE_CODE}Details: {err}"
                ) from err
        raise AuthError(_AUTH_FAILURE_HINT_INTERACTIVE_NO_CLIENT)

    raise AuthError(_AUTH_FAILURE_HINT_NONINTERACTIVE)

napt.upload.graph

Microsoft Graph API and Azure Blob Storage client for Intune Win32 app upload.

Implements the full upload flow for a Win32 LOB app:

1. Create Win32 app record in Intune (POST mobileApps)
2. Create a content version (POST contentVersions)
3. Create a file entry and wait for SAS URI (POST files + polling)
4. Upload encrypted payload to Azure Blob Storage (PUT blocks + block list)
5. Commit the uploaded file with encryption metadata (POST commit + polling)
6. Set the committed content version on the app (PATCH mobileApps)

Also provides app queries used for reconciliation (list_mobile_apps, get_mobile_app, update_win32_app) and group-based assignment plumbing (resolve_group_id, get_app_assignments, build_group_assignment, assign_app) used by deployment promotion.

All functions take an access_token as the first argument. Obtain one via napt.upload.auth.get_access_token().

Graph calls retry transient failures — HTTP 429 (honoring Retry-After), transient server errors, and connection drops — with bounded exponential backoff before raising. Resource-creating POSTs retry only unambiguous throttling responses, so a lost reply to a processed create is never resubmitted as a duplicate. Azure Blob PUTs carry their own retry tuned for SAS-propagation 403s.

Example

Full upload flow:

from pathlib import Path
from napt.upload.auth import get_access_token
from napt.upload.graph import (
    create_win32_app, create_content_version,
    create_content_version_file, upload_to_azure_blob,
    commit_content_version_file, commit_content_version,
)
from napt.upload.intunewin import parse_intunewin

token = get_access_token()
metadata = parse_intunewin(Path("packages/napt-chrome/144.0.7559.110/Invoke-AppDeployToolkit.intunewin"))
app_id = create_win32_app(token, app_metadata)
cv_id = create_content_version(token, app_id)
file_id, sas_uri = create_content_version_file(token, app_id, cv_id, metadata)
upload_to_azure_blob(sas_uri, Path("/tmp/IntunePackage.intunewin"))
commit_content_version_file(token, app_id, cv_id, file_id, metadata)
commit_content_version(token, app_id, cv_id)

resolve_group_id

resolve_group_id(access_token: str, group: str) -> str

Resolves an Entra ID group name or object ID to an object ID.

Values that already look like object IDs (GUIDs) pass through without a Graph call. Names are looked up by exact displayName match, which requires the Group.Read.All application permission.

Parameters:

Name	Type	Description	Default
`access_token`	`str`	Bearer token for Graph API.	required
`group`	`str`	Group displayName or object ID (GUID).	required

Returns:

Type	Description
`str`	The group's object ID.

Raises:

Type	Description
`AuthError`	On 401 or 403 (check Group.Read.All permission).
`ConfigError`	If no group or more than one group matches the name.
`NetworkError`	On 5xx or connection error.

Source code in napt/upload/graph.py

def resolve_group_id(access_token: str, group: str) -> str:
    """Resolves an Entra ID group name or object ID to an object ID.

    Values that already look like object IDs (GUIDs) pass through without
    a Graph call. Names are looked up by exact displayName match, which
    requires the Group.Read.All application permission.

    Args:
        access_token: Bearer token for Graph API.
        group: Group displayName or object ID (GUID).

    Returns:
        The group's object ID.

    Raises:
        AuthError: On 401 or 403 (check Group.Read.All permission).
        ConfigError: If no group or more than one group matches the name.
        NetworkError: On 5xx or connection error.

    """
    if _GUID_RE.match(group):
        return group

    escaped = group.replace("'", "''")
    url = (
        f"{GRAPH_BASE}/groups"
        f"?$filter=displayName eq '{escaped}'&$select=id,displayName"
    )
    body = _graph_request(
        "GET", url, "resolve_group_id", headers=_auth_headers(access_token)
    )
    matches: list[dict] = body.get("value", [])

    if not matches:
        raise ConfigError(
            f"No Entra ID group found with displayName '{group}'. "
            "Check the name, or use the group's object ID instead."
        )
    if len(matches) > 1:
        ids = ", ".join(m["id"] for m in matches)
        raise ConfigError(
            f"Multiple Entra ID groups share the displayName '{group}' "
            f"({ids}). Use the object ID of the intended group instead."
        )
    return matches[0]["id"]

resolve_assignment_target

resolve_assignment_target(
    access_token: str, group: str, group_id_cache: dict[str, str] | None = None
) -> dict

Resolves a deployment group entry to an assignment target dict.

The reserved names "All Users" and "All Devices" map to Intune's built-in virtual targets; anything else resolves to an Entra ID group target via resolve_group_id.

Parameters:

Name	Type	Description	Default
`access_token`	`str`	Bearer token for Graph API.	required
`group`	`str`	Group displayName, object ID, or reserved virtual name.	required
`group_id_cache`	`dict[str, str] \| None`	Optional cache of name to object ID, shared across calls to avoid repeated lookups.	`None`

Returns:

Type	Description
`dict`	An assignment target dict for use with build_assignment.

Raises:

Type	Description
`AuthError`	On 401 or 403 (check Group.Read.All permission).
`ConfigError`	If no group or more than one group matches a name.
`NetworkError`	On 5xx or connection error.

Source code in napt/upload/graph.py

def resolve_assignment_target(
    access_token: str,
    group: str,
    group_id_cache: dict[str, str] | None = None,
) -> dict:
    """Resolves a deployment group entry to an assignment target dict.

    The reserved names "All Users" and "All Devices" map to Intune's
    built-in virtual targets; anything else resolves to an Entra ID
    group target via resolve_group_id.

    Args:
        access_token: Bearer token for Graph API.
        group: Group displayName, object ID, or reserved virtual name.
        group_id_cache: Optional cache of name to object ID, shared
            across calls to avoid repeated lookups.

    Returns:
        An assignment target dict for use with build_assignment.

    Raises:
        AuthError: On 401 or 403 (check Group.Read.All permission).
        ConfigError: If no group or more than one group matches a name.
        NetworkError: On 5xx or connection error.

    """
    if group in VIRTUAL_TARGETS:
        return dict(VIRTUAL_TARGETS[group])
    if group_id_cache is None:
        group_id_cache = {}
    if group not in group_id_cache:
        group_id_cache[group] = resolve_group_id(access_token, group)
    return {
        "@odata.type": "#microsoft.graph.groupAssignmentTarget",
        "groupId": group_id_cache[group],
    }

get_app_assignments

get_app_assignments(access_token: str, app_id: str) -> list[dict]

Gets the current assignments of a mobile app.

Parameters:

Name	Type	Description	Default
`access_token`	`str`	Bearer token for Graph API.	required
`app_id`	`str`	Graph API object ID of the app.	required

Returns:

Type	Description
`list[dict]`	A list of mobileAppAssignment dicts (empty when unassigned).

Raises:

Type	Description
`AuthError`	On 401 or 403.
`NetworkError`	On 5xx or connection error.

Source code in napt/upload/graph.py

def get_app_assignments(access_token: str, app_id: str) -> list[dict]:
    """Gets the current assignments of a mobile app.

    Args:
        access_token: Bearer token for Graph API.
        app_id: Graph API object ID of the app.

    Returns:
        A list of mobileAppAssignment dicts (empty when unassigned).

    Raises:
        AuthError: On 401 or 403.
        NetworkError: On 5xx or connection error.

    """
    url = f"{GRAPH_BASE}/deviceAppManagement/mobileApps/{app_id}/assignments"
    body = _graph_request(
        "GET", url, "get_app_assignments", headers=_auth_headers(access_token)
    )
    return body.get("value", [])

build_assignment

build_assignment(target: dict, intent: str) -> dict

Builds a mobileAppAssignment payload for a resolved target.

Parameters:

Name	Type	Description	Default
`target`	`dict`	An assignment target dict (group or virtual target).	required
`intent`	`str`	Assignment intent, "available" or "required".	required

Returns:

Type	Description
`dict`	A mobileAppAssignment dict for use with assign_app.

Source code in napt/upload/graph.py

def build_assignment(target: dict, intent: str) -> dict:
    """Builds a mobileAppAssignment payload for a resolved target.

    Args:
        target: An assignment target dict (group or virtual target).
        intent: Assignment intent, "available" or "required".

    Returns:
        A mobileAppAssignment dict for use with assign_app.

    """
    return {
        "@odata.type": "#microsoft.graph.mobileAppAssignment",
        "intent": intent,
        "target": target,
    }

build_group_assignment

build_group_assignment(group_id: str, intent: str) -> dict

Builds a mobileAppAssignment payload targeting one Entra ID group.

Parameters:

Name	Type	Description	Default
`group_id`	`str`	Object ID of the target group.	required
`intent`	`str`	Assignment intent, "available" or "required".	required

Returns:

Type	Description
`dict`	A mobileAppAssignment dict for use with assign_app.

Source code in napt/upload/graph.py

def build_group_assignment(group_id: str, intent: str) -> dict:
    """Builds a mobileAppAssignment payload targeting one Entra ID group.

    Args:
        group_id: Object ID of the target group.
        intent: Assignment intent, "available" or "required".

    Returns:
        A mobileAppAssignment dict for use with assign_app.

    """
    return build_assignment(
        {
            "@odata.type": "#microsoft.graph.groupAssignmentTarget",
            "groupId": group_id,
        },
        intent,
    )

assign_app

assign_app(access_token: str, app_id: str, assignments: list[dict]) -> None

Sets a mobile app's assignments.

The assign action replaces the app's entire assignment set. Callers that intend to preserve existing assignments must read them first with get_app_assignments and include them in the new list.

Parameters:

Name	Type	Description	Default
`access_token`	`str`	Bearer token for Graph API.	required
`app_id`	`str`	Graph API object ID of the app.	required
`assignments`	`list[dict]`	Complete list of mobileAppAssignment dicts to apply.	required

Raises:

Type	Description
`AuthError`	On 401 or 403.
`ConfigError`	On 400 (invalid assignment payload).
`NetworkError`	On 5xx or connection error.

Source code in napt/upload/graph.py

def assign_app(access_token: str, app_id: str, assignments: list[dict]) -> None:
    """Sets a mobile app's assignments.

    The assign action replaces the app's entire assignment set. Callers
    that intend to preserve existing assignments must read them first with
    get_app_assignments and include them in the new list.

    Args:
        access_token: Bearer token for Graph API.
        app_id: Graph API object ID of the app.
        assignments: Complete list of mobileAppAssignment dicts to apply.

    Raises:
        AuthError: On 401 or 403.
        ConfigError: On 400 (invalid assignment payload).
        NetworkError: On 5xx or connection error.

    """
    url = f"{GRAPH_BASE}/deviceAppManagement/mobileApps/{app_id}/assign"
    body = {"mobileAppAssignments": assignments}
    _graph_request(
        "POST", url, "assign_app", headers=_json_headers(access_token), json=body
    )

list_mobile_apps

list_mobile_apps(access_token: str) -> list[dict]

Lists all mobile apps in the tenant with id, displayName, and notes.

Follows @odata.nextLink pagination until the collection is exhausted.

Parameters:

Name	Type	Description	Default
`access_token`	`str`	Bearer token for Graph API.	required

Returns:

Type	Description
`list[dict]`	A list of app dicts, each with at least "id", "displayName", and "notes" keys.

Raises:

Type	Description
`AuthError`	On 401 or 403.
`NetworkError`	On 5xx or connection error.

Source code in napt/upload/graph.py

def list_mobile_apps(access_token: str) -> list[dict]:
    """Lists all mobile apps in the tenant with id, displayName, and notes.

    Follows @odata.nextLink pagination until the collection is exhausted.

    Args:
        access_token: Bearer token for Graph API.

    Returns:
        A list of app dicts, each with at least "id", "displayName", and
            "notes" keys.

    Raises:
        AuthError: On 401 or 403.
        NetworkError: On 5xx or connection error.

    """
    url: str | None = (
        f"{GRAPH_BASE}/deviceAppManagement/mobileApps" "?$select=id,displayName,notes"
    )
    apps: list[dict] = []
    while url:
        body = _graph_request(
            "GET", url, "list_mobile_apps", headers=_auth_headers(access_token)
        )
        apps.extend(body.get("value", []))
        url = body.get("@odata.nextLink")
    return apps

delete_mobile_app

delete_mobile_app(access_token: str, app_id: str) -> None

Deletes a mobile app from Intune.

A 404 is tolerated — the app being already gone is the desired end state, so retried deletions stay idempotent.

Parameters:

Name	Type	Description	Default
`access_token`	`str`	Bearer token for Graph API.	required
`app_id`	`str`	Graph API object ID of the app to delete.	required

Raises:

Type	Description
`AuthError`	On 401 or 403.
`NetworkError`	On 5xx or connection error.

Source code in napt/upload/graph.py

def delete_mobile_app(access_token: str, app_id: str) -> None:
    """Deletes a mobile app from Intune.

    A 404 is tolerated — the app being already gone is the desired end
    state, so retried deletions stay idempotent.

    Args:
        access_token: Bearer token for Graph API.
        app_id: Graph API object ID of the app to delete.

    Raises:
        AuthError: On 401 or 403.
        NetworkError: On 5xx or connection error.

    """
    url = f"{GRAPH_BASE}/deviceAppManagement/mobileApps/{app_id}"
    _graph_request(
        "DELETE",
        url,
        "delete_mobile_app",
        headers=_auth_headers(access_token),
        ok_statuses=(404,),
    )

get_mobile_app

get_mobile_app(access_token: str, app_id: str) -> dict

Gets one mobile app's full object by Graph API ID.

Used to read subtype fields that $select on the collection cannot reliably return, such as win32LobApp.committedContentVersion.

Parameters:

Name	Type	Description	Default
`access_token`	`str`	Bearer token for Graph API.	required
`app_id`	`str`	Graph API object ID of the app.	required

Returns:

Type	Description
`dict`	The full app object dict.

Raises:

Type	Description
`AuthError`	On 401 or 403.
`NetworkError`	On 5xx or connection error.

Source code in napt/upload/graph.py

def get_mobile_app(access_token: str, app_id: str) -> dict:
    """Gets one mobile app's full object by Graph API ID.

    Used to read subtype fields that $select on the collection cannot
    reliably return, such as win32LobApp.committedContentVersion.

    Args:
        access_token: Bearer token for Graph API.
        app_id: Graph API object ID of the app.

    Returns:
        The full app object dict.

    Raises:
        AuthError: On 401 or 403.
        NetworkError: On 5xx or connection error.

    """
    url = f"{GRAPH_BASE}/deviceAppManagement/mobileApps/{app_id}"
    return _graph_request(
        "GET", url, "get_mobile_app", headers=_auth_headers(access_token)
    )

create_win32_app

create_win32_app(access_token: str, app_metadata: dict) -> str

Creates a new Win32 LOB app record in Intune.

Parameters:

Name	Type	Description	Default
`access_token`	`str`	Bearer token for Graph API.	required
`app_metadata`	`dict`	Win32LobApp JSON payload (display name, install commands, detection rules, etc.).	required

Returns:

Type	Description
`str`	The Graph API object ID of the newly created app.

Raises:

Type	Description
`AuthError`	On 401 or 403.
`ConfigError`	On 400 (invalid metadata).
`NetworkError`	On 5xx or connection error.

Source code in napt/upload/graph.py

def create_win32_app(access_token: str, app_metadata: dict) -> str:
    """Creates a new Win32 LOB app record in Intune.

    Args:
        access_token: Bearer token for Graph API.
        app_metadata: Win32LobApp JSON payload (display name, install
            commands, detection rules, etc.).

    Returns:
        The Graph API object ID of the newly created app.

    Raises:
        AuthError: On 401 or 403.
        ConfigError: On 400 (invalid metadata).
        NetworkError: On 5xx or connection error.

    """
    url = f"{GRAPH_BASE}/deviceAppManagement/mobileApps"
    body = _graph_request(
        "POST",
        url,
        "create_win32_app",
        headers=_json_headers(access_token),
        json=app_metadata,
        idempotent=False,
    )
    return body["id"]

update_win32_app

update_win32_app(access_token: str, app_id: str, app_metadata: dict) -> None

Updates an existing Win32 LOB app record's metadata in Intune.

Parameters:

Name	Type	Description	Default
`access_token`	`str`	Bearer token for Graph API.	required
`app_id`	`str`	Graph API object ID of the app to update.	required
`app_metadata`	`dict`	Win32LobApp JSON payload to apply.	required

Raises:

Type	Description
`AuthError`	On 401 or 403.
`ConfigError`	On 400 (invalid metadata).
`NetworkError`	On 5xx or connection error.

Source code in napt/upload/graph.py

def update_win32_app(access_token: str, app_id: str, app_metadata: dict) -> None:
    """Updates an existing Win32 LOB app record's metadata in Intune.

    Args:
        access_token: Bearer token for Graph API.
        app_id: Graph API object ID of the app to update.
        app_metadata: Win32LobApp JSON payload to apply.

    Raises:
        AuthError: On 401 or 403.
        ConfigError: On 400 (invalid metadata).
        NetworkError: On 5xx or connection error.

    """
    url = f"{GRAPH_BASE}/deviceAppManagement/mobileApps/{app_id}"
    _graph_request(
        "PATCH",
        url,
        "update_win32_app",
        headers=_json_headers(access_token),
        json=app_metadata,
    )

create_content_version

create_content_version(access_token: str, app_id: str) -> str

Creates a new content version for a Win32 app.

Parameters:

Name	Type	Description	Default
`access_token`	`str`	Bearer token for Graph API.	required
`app_id`	`str`	Graph API object ID of the Win32 app.	required

Returns:

Type	Description
`str`	The content version ID string.

Raises:

Type	Description
`AuthError`	On 401 or 403.
`NetworkError`	On 5xx or connection error.

Source code in napt/upload/graph.py

def create_content_version(access_token: str, app_id: str) -> str:
    """Creates a new content version for a Win32 app.

    Args:
        access_token: Bearer token for Graph API.
        app_id: Graph API object ID of the Win32 app.

    Returns:
        The content version ID string.

    Raises:
        AuthError: On 401 or 403.
        NetworkError: On 5xx or connection error.

    """
    url = (
        f"{GRAPH_BASE}/deviceAppManagement/mobileApps/{app_id}"
        f"/microsoft.graph.win32LobApp/contentVersions"
    )
    body = _graph_request(
        "POST",
        url,
        "create_content_version",
        headers=_json_headers(access_token),
        json={},
        idempotent=False,
    )
    return body["id"]

create_content_version_file

create_content_version_file(
    access_token: str, app_id: str, cv_id: str, metadata: IntunewinMetadata
) -> tuple[str, str]

Creates a file entry for a content version and waits for the SAS URI.

Posts the file size information to Graph API, then polls until Azure Storage has provisioned a SAS URI for the upload.

Parameters:

Name	Type	Description	Default
`access_token`	`str`	Bearer token for Graph API.	required
`app_id`	`str`	Graph API object ID of the Win32 app.	required
`cv_id`	`str`	Content version ID from create_content_version.	required
`metadata`	`IntunewinMetadata`	Parsed .intunewin metadata (provides file sizes).	required

Returns:

Type	Description
`tuple[str, str]`	A tuple of (file_id, sas_uri) where sas_uri is the Azure Blob Storage SAS URI to upload the encrypted payload to.

Raises:

Type	Description
`AuthError`	On 401 or 403.
`NetworkError`	On 5xx, connection error, or upload state error.

Source code in napt/upload/graph.py

def create_content_version_file(
    access_token: str,
    app_id: str,
    cv_id: str,
    metadata: IntunewinMetadata,
) -> tuple[str, str]:
    """Creates a file entry for a content version and waits for the SAS URI.

    Posts the file size information to Graph API, then polls until Azure
    Storage has provisioned a SAS URI for the upload.

    Args:
        access_token: Bearer token for Graph API.
        app_id: Graph API object ID of the Win32 app.
        cv_id: Content version ID from create_content_version.
        metadata: Parsed .intunewin metadata (provides file sizes).

    Returns:
        A tuple of (file_id, sas_uri) where sas_uri is the Azure Blob
            Storage SAS URI to upload the encrypted payload to.

    Raises:
        AuthError: On 401 or 403.
        NetworkError: On 5xx, connection error, or upload state error.

    """
    base_url = (
        f"{GRAPH_BASE}/deviceAppManagement/mobileApps/{app_id}"
        f"/microsoft.graph.win32LobApp/contentVersions/{cv_id}/files"
    )
    body = {
        "@odata.type": "#microsoft.graph.mobileAppContentFile",
        "name": metadata.encrypted_file_name,
        "size": metadata.unencrypted_content_size,
        "sizeEncrypted": metadata.encrypted_file_size,
        "manifest": None,
        "isDependency": False,
    }
    file_body = _graph_request(
        "POST",
        base_url,
        "create_content_version_file",
        headers=_json_headers(access_token),
        json=body,
        idempotent=False,
    )
    file_id: str = file_body["id"]

    poll_url = f"{base_url}/{file_id}"
    data = _poll(
        access_token,
        poll_url,
        success_state="azureStorageUriRequestSuccess",
        context="create_content_version_file (poll SAS URI)",
    )
    return file_id, data["azureStorageUri"]

upload_to_azure_blob

upload_to_azure_blob(sas_uri: str, encrypted_payload_path: Path) -> None

Uploads the encrypted payload to Azure Blob Storage using block blobs.

Splits the file into CHUNK_SIZE chunks, uploads each as a block with a base64-encoded block ID, then commits the block list. Prints an inline progress percentage as each chunk completes. Transient per-request failures (including 403 from a not-yet-propagated SAS URI) are retried with backoff.

Parameters:

Name	Type	Description	Default
`sas_uri`	`str`	Azure Blob Storage SAS URI from create_content_version_file.	required
`encrypted_payload_path`	`Path`	Path to the extracted encrypted payload file (IntunePackage.intunewin from inside the .intunewin ZIP).	required

Raises:

Type	Description
`NetworkError`	If any block upload or the block list commit fails after retries.

Source code in napt/upload/graph.py

def upload_to_azure_blob(
    sas_uri: str,
    encrypted_payload_path: Path,
) -> None:
    """Uploads the encrypted payload to Azure Blob Storage using block blobs.

    Splits the file into CHUNK_SIZE chunks, uploads each as a block with a
    base64-encoded block ID, then commits the block list. Prints an inline
    progress percentage as each chunk completes. Transient per-request
    failures (including 403 from a not-yet-propagated SAS URI) are retried
    with backoff.

    Args:
        sas_uri: Azure Blob Storage SAS URI from create_content_version_file.
        encrypted_payload_path: Path to the extracted encrypted payload file
            (IntunePackage.intunewin from inside the .intunewin ZIP).

    Raises:
        NetworkError: If any block upload or the block list commit fails
            after retries.

    """
    from napt.logging import get_global_logger

    logger = get_global_logger()

    block_ids: list[str] = []
    total_bytes = encrypted_payload_path.stat().st_size
    bytes_uploaded = 0
    last_percent = -1

    started_at = time.time()
    with open(encrypted_payload_path, "rb") as fh:
        block_index = 0
        while True:
            chunk = fh.read(CHUNK_SIZE)
            if not chunk:
                break

            # Block ID: base64(zero-padded 5-digit decimal index)
            block_id = base64.b64encode(str(block_index).zfill(5).encode()).decode()
            block_ids.append(block_id)

            put_url = f"{sas_uri}&comp=block&blockid={block_id}"
            _blob_put_with_retry(
                put_url,
                chunk,
                headers={
                    "x-ms-blob-type": "BlockBlob",
                    "Content-Length": str(len(chunk)),
                },
                context=f"Azure Blob block upload failed (block {block_index})",
            )

            bytes_uploaded += len(chunk)
            if total_bytes:
                pct = int(bytes_uploaded * 100 / total_bytes)
                if pct != last_percent:
                    logger.progress("UPLOAD", f"{pct}%")
                    last_percent = pct

            block_index += 1

    # Commit all blocks by submitting the block list
    block_list_xml = (
        '<?xml version="1.0" encoding="utf-8"?>\n'
        "<BlockList>\n"
        + "".join(f"  <Latest>{bid}</Latest>\n" for bid in block_ids)
        + "</BlockList>"
    )
    commit_url = f"{sas_uri}&comp=blocklist"
    _blob_put_with_retry(
        commit_url,
        block_list_xml.encode("utf-8"),
        headers={"Content-Type": "application/xml"},
        context="Azure Blob block list commit failed",
        timeout=60,
    )

    elapsed = time.time() - started_at
    speed_mb = (bytes_uploaded / (1024 * 1024)) / elapsed if elapsed > 0 else 0
    size_mb = bytes_uploaded / (1024 * 1024)
    logger.info(
        "UPLOAD",
        f"Complete: {encrypted_payload_path.name} ({size_mb:.1f} MB) "
        f"in {elapsed:.1f}s at {speed_mb:.1f} MB/s",
    )

commit_content_version_file

commit_content_version_file(
    access_token: str,
    app_id: str,
    cv_id: str,
    file_id: str,
    metadata: IntunewinMetadata,
) -> None

Commits the uploaded file with encryption metadata, then waits for confirmation.

Sends the encryption key, MAC, IV, and digest to Graph API, then polls until Intune confirms the file is committed.

Parameters:

Name	Type	Description	Default
`access_token`	`str`	Bearer token for Graph API.	required
`app_id`	`str`	Graph API object ID of the Win32 app.	required
`cv_id`	`str`	Content version ID.	required
`file_id`	`str`	File entry ID from create_content_version_file.	required
`metadata`	`IntunewinMetadata`	Parsed .intunewin metadata (provides all encryption fields).	required

Raises:

Type	Description
`AuthError`	On 401 or 403.
`NetworkError`	On 5xx, connection error, or if commit times out.

Note

Graph returns 200 (not 201) for the commit POST.

Source code in napt/upload/graph.py

def commit_content_version_file(
    access_token: str,
    app_id: str,
    cv_id: str,
    file_id: str,
    metadata: IntunewinMetadata,
) -> None:
    """Commits the uploaded file with encryption metadata, then waits for confirmation.

    Sends the encryption key, MAC, IV, and digest to Graph API, then polls
    until Intune confirms the file is committed.

    Args:
        access_token: Bearer token for Graph API.
        app_id: Graph API object ID of the Win32 app.
        cv_id: Content version ID.
        file_id: File entry ID from create_content_version_file.
        metadata: Parsed .intunewin metadata (provides all encryption fields).

    Raises:
        AuthError: On 401 or 403.
        NetworkError: On 5xx, connection error, or if commit times out.

    Note:
        Graph returns 200 (not 201) for the commit POST.

    """
    commit_url = (
        f"{GRAPH_BASE}/deviceAppManagement/mobileApps/{app_id}"
        f"/microsoft.graph.win32LobApp/contentVersions/{cv_id}"
        f"/files/{file_id}/commit"
    )
    body = {
        "fileEncryptionInfo": {
            "encryptionKey": metadata.encryption_key,
            "macKey": metadata.mac_key,
            "initializationVector": metadata.init_vector,
            "mac": metadata.mac,
            "profileIdentifier": metadata.profile_identifier,
            "fileDigest": metadata.file_digest,
            "fileDigestAlgorithm": metadata.file_digest_algorithm,
        }
    }
    _graph_request(
        "POST",
        commit_url,
        "commit_content_version_file",
        headers=_json_headers(access_token),
        json=body,
    )

    poll_url = (
        f"{GRAPH_BASE}/deviceAppManagement/mobileApps/{app_id}"
        f"/microsoft.graph.win32LobApp/contentVersions/{cv_id}/files/{file_id}"
    )
    _poll(
        access_token,
        poll_url,
        success_state="commitFileSuccess",
        context="commit_content_version_file (poll commit)",
    )

commit_content_version

commit_content_version(access_token: str, app_id: str, cv_id: str) -> None

Sets the committed content version on the Win32 app.

This is the final step — after calling this, the app is fully published in Intune and available for assignment.

Parameters:

Name	Type	Description	Default
`access_token`	`str`	Bearer token for Graph API.	required
`app_id`	`str`	Graph API object ID of the Win32 app.	required
`cv_id`	`str`	Content version ID to mark as committed.	required

Raises:

Type	Description
`AuthError`	On 401 or 403.
`NetworkError`	On 5xx or connection error.

Source code in napt/upload/graph.py

def commit_content_version(access_token: str, app_id: str, cv_id: str) -> None:
    """Sets the committed content version on the Win32 app.

    This is the final step — after calling this, the app is fully published
    in Intune and available for assignment.

    Args:
        access_token: Bearer token for Graph API.
        app_id: Graph API object ID of the Win32 app.
        cv_id: Content version ID to mark as committed.

    Raises:
        AuthError: On 401 or 403.
        NetworkError: On 5xx or connection error.

    """
    url = f"{GRAPH_BASE}/deviceAppManagement/mobileApps/{app_id}"
    body = {
        "@odata.type": WIN32_LOB_APP_TYPE,
        "committedContentVersion": cv_id,
    }
    _graph_request(
        "PATCH",
        url,
        "commit_content_version",
        headers=_json_headers(access_token),
        json=body,
    )

upload

napt.upload.manager

upload_package

napt.upload.intunewin

IntunewinMetadata dataclass

parse_intunewin

extract_encrypted_payload

napt.upload.auth

get_credential

get_access_token

napt.upload.graph

resolve_group_id

resolve_assignment_target

get_app_assignments

build_assignment

build_group_assignment

assign_app

list_mobile_apps

delete_mobile_app

get_mobile_app

create_win32_app

update_win32_app

create_content_version

create_content_version_file

upload_to_azure_blob

commit_content_version_file

commit_content_version

IntunewinMetadata `dataclass`