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) -> 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.

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

Parameters:

Name	Type	Description	Default
recipe_path	Path	Path to the recipe YAML file.	required

Returns:

Type	Description
UploadResult	Upload result including the Intune app ID, app name, version, and package path.

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.

Example

Upload and print the resulting Intune app ID:

from pathlib import Path
from napt.upload import upload_package

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

Source code in napt/upload/manager.py

def upload_package(recipe_path: Path) -> 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.

    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

    Args:
        recipe_path: Path to the recipe YAML file.

    Returns:
        Upload result including the Intune app ID, app name, version, and
            package path.

    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.

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

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

    """
    logger = get_global_logger()

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

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

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

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

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

    # Step 4: Create Intune app record and content version
    logger.step(4, 6, f"Creating Intune app record for '{app_name}' {version}...")
    app_metadata = _build_app_metadata(config, recipe_path, version, package_path)
    intune_app_id = create_win32_app(access_token, app_metadata)
    logger.info("UPLOAD", f"Created Intune app: {intune_app_id}")

    cv_id = create_content_version(access_token, intune_app_id)
    logger.verbose("UPLOAD", f"Content version: {cv_id}")

    file_id, sas_uri = create_content_version_file(
        access_token, intune_app_id, cv_id, intunewin_metadata
    )
    logger.verbose("UPLOAD", f"File entry: {file_id}")

    # Step 5: Upload encrypted payload to Azure Blob Storage
    logger.step(5, 6, "Uploading to Azure Blob Storage...")
    with tempfile.TemporaryDirectory() as tmp_dir:
        payload_path = extract_encrypted_payload(package_path, Path(tmp_dir))
        upload_to_azure_blob(sas_uri, payload_path)

    # Step 6: Commit
    logger.step(6, 6, "Committing content version...")
    commit_content_version_file(
        access_token, intune_app_id, cv_id, file_id, intunewin_metadata
    )
    commit_content_version(access_token, intune_app_id, cv_id)

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

    return UploadResult(
        app_id=app_id,
        app_name=app_name,
        version=version,
        intune_app_id=intune_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

1. 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.).
2. 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.
3. 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)

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

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)

create_win32_app

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

Create 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:
    """Create 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"
    resp = requests.post(
        url, headers=_json_headers(access_token), json=app_metadata, timeout=30
    )
    body = _check_response(resp, "create_win32_app")
    return body["id"]

create_content_version

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

Create 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:
    """Create 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"
    )
    resp = requests.post(url, headers=_json_headers(access_token), json={}, timeout=30)
    body = _check_response(resp, "create_content_version")
    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]

Create a file entry for a content version and wait 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]:
    """Create a file entry for a content version and wait 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,
    }
    resp = requests.post(
        base_url, headers=_json_headers(access_token), json=body, timeout=30
    )
    file_body = _check_response(resp, "create_content_version_file")
    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

Upload 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.

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.

Source code in napt/upload/graph.py

def upload_to_azure_blob(
    sas_uri: str,
    encrypted_payload_path: Path,
) -> None:
    """Upload 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.

    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.

    """
    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}"
            resp = requests.put(
                put_url,
                data=chunk,
                headers={
                    "x-ms-blob-type": "BlockBlob",
                    "Content-Length": str(len(chunk)),
                },
                timeout=300,
            )
            if not resp.ok:
                raise NetworkError(
                    f"Azure Blob block upload failed (block {block_index}): "
                    f"HTTP {resp.status_code}\n{resp.text}"
                )

            bytes_uploaded += len(chunk)
            if total_bytes:
                pct = int(bytes_uploaded * 100 / total_bytes)
                if pct != last_percent:
                    # TODO: Route progress output through the logger once a
                    # logger.progress() method is added. Bare print bypasses
                    # SilentLogger in library usage.
                    print(f"upload progress: {pct}%", end="\r")
                    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"
    resp = requests.put(
        commit_url,
        data=block_list_xml.encode("utf-8"),
        headers={"Content-Type": "application/xml"},
        timeout=60,
    )
    if not resp.ok:
        raise NetworkError(
            f"Azure Blob block list commit failed: HTTP {resp.status_code}\n{resp.text}"
        )

    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

Commit the uploaded file with encryption metadata, then wait 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.

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:
    """Commit the uploaded file with encryption metadata, then wait 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.

    """
    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,
        }
    }
    resp = requests.post(
        commit_url, headers=_json_headers(access_token), json=body, timeout=30
    )
    # Graph API returns 200 for this call
    _check_response(resp, "commit_content_version_file")

    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

Set 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:
    """Set 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,
    }
    resp = requests.patch(
        url, headers=_json_headers(access_token), json=body, timeout=30
    )
    _check_response(resp, "commit_content_version")