MOVEit Automation import sample

pwsh ./runimport.ps1 /path/to/task-export.xml

python3 -m pip install PyJWT requests cryptography

python3 runimport.py /path/to/task-export.xml

#requires -Version 7.0

param(
    [Parameter(Mandatory = $true, Position = 0)]
    [string]$FilePath
)

# --- Static config -----------------------------------------------------
$BaseUrl        = "https://api.<env>.mft.progress.com"
$PrivateKeyFile = ""   # Path to the EC private key PEM file.
$TenantId       = ""   # Automate MFT tenant ID. You can get it from Plans and Usage page in the Automate MFT UI.
$Kid            = ""   # Key ID of the EC private key. You can get it from API Keys page in the Automate MFT UI.
$TargetFolderId = ""   # Automate MFT target folder ID. Leave empty to import into All Tasks. You can get the value by selecting the folder in the Tasks page and get the value of the folderIds query parameter in the URL.
# ----------------------------------------------------------------------

function Get-HttpStatusCode {
    param($ErrorRecord)

    if ($ErrorRecord.Exception.StatusCode) {
        return [int]$ErrorRecord.Exception.StatusCode
    }

    if ($ErrorRecord.Exception.Response -and $ErrorRecord.Exception.Response.StatusCode) {
        return [int]$ErrorRecord.Exception.Response.StatusCode
    }

    return $null
}

function Retry-On429 {
    param(
        [Parameter(Mandatory = $true)]
        [ScriptBlock]$Script,

        [int]$MaxRetries = 3,

        [int]$RetryDelaySeconds = 5
    )

    $retryCount = 0

    while ($true) {
        try {
            return & $Script
        }
        catch {
            if ((Get-HttpStatusCode $_) -ne 429) {
                throw
            }

            $retryCount++
            if ($retryCount -ge $MaxRetries) {
                throw
            }

            Write-Host "Received HTTP 429. Retrying in $RetryDelaySeconds seconds... (attempt $retryCount/$MaxRetries)"
            Start-Sleep -Seconds $RetryDelaySeconds
        }
    }
}

function ConvertTo-Base64Url {
    param(
        [Parameter(Mandatory = $true)]
        [byte[]]$Bytes
    )

    return [Convert]::ToBase64String($Bytes).TrimEnd('=').Replace('+', '-').Replace('/', '_')
}

function ConvertFrom-JsonObjectToBase64Url {
    param(
        [Parameter(Mandatory = $true)]
        [hashtable]$Object
    )

    $json = $Object | ConvertTo-Json -Compress -Depth 10
    $bytes = [System.Text.Encoding]::UTF8.GetBytes($json)
    return ConvertTo-Base64Url -Bytes $bytes
}

function Create-JWTToken {
    param(
        [Parameter(Mandatory = $true)]
        [string]$PrivateKeyPem,

        [Parameter(Mandatory = $true)]
        [string]$KeyId,

        [Parameter(Mandatory = $true)]
        [string]$Tenant,

        [Parameter(Mandatory = $true)]
        [string]$ApiBaseUrl
    )

    $now = [DateTimeOffset]::UtcNow.ToUnixTimeSeconds()

    $header = @{
        alg = "ES256"
        kid = $KeyId
        typ = "JWT"
    }

    $payload = @{
        aud = "$ApiBaseUrl/v1/oauth/token"
        jti = [guid]::NewGuid().ToString()
        iat = $now
        nbf = $now
        exp = $now + 240
        iss = $Tenant
        sub = $Tenant
    }

    $encodedHeader  = ConvertFrom-JsonObjectToBase64Url -Object $header
    $encodedPayload = ConvertFrom-JsonObjectToBase64Url -Object $payload
    $signingInput   = "$encodedHeader.$encodedPayload"
    $signingBytes   = [System.Text.Encoding]::ASCII.GetBytes($signingInput)

    $ecdsa = [System.Security.Cryptography.ECDsa]::Create()

    try {
        $ecdsa.ImportFromPem($PrivateKeyPem)

        $signature = $ecdsa.SignData(
            $signingBytes,
            [System.Security.Cryptography.HashAlgorithmName]::SHA256,
            [System.Security.Cryptography.DSASignatureFormat]::IeeeP1363FixedFieldConcatenation
        )

        return "$signingInput.$(ConvertTo-Base64Url -Bytes $signature)"
    }
    finally {
        $ecdsa.Dispose()
    }
}

function Get-AccessToken {
    param(
        [Parameter(Mandatory = $true)]
        [string]$JWT,

        [Parameter(Mandatory = $true)]
        [string]$ApiBaseUrl
    )

    $body = @{
        grant_type            = "client_credentials"
        client_assertion_type = "urn:ietf:params:oauth:client-assertion-type:jwt-bearer"
        client_assertion      = $JWT
    } | ConvertTo-Json -Compress

    $response = Retry-On429 -Script {
        Invoke-RestMethod `
            -Method Post `
            -Uri "$ApiBaseUrl/v1/oauth/token" `
            -Headers @{ "Content-Type" = "application/json" } `
            -Body $body `
            -TimeoutSec 30 `
            -ErrorAction Stop
    }

    if ([string]::IsNullOrWhiteSpace($response.access_token)) {
        throw "The OAuth token response did not contain an access_token."
    }

    return $response.access_token
}

function Read-MiaPassphrase {
    $securePassphrase = Read-Host -Prompt "MIA export passphrase (press Enter if none)" -AsSecureString
    $credential = [System.Net.NetworkCredential]::new("", $securePassphrase)
    return $credential.Password
}

function Invoke-MiaImport {
    param(
        [Parameter(Mandatory = $true)]
        [ValidateSet("preview", "apply")]
        [string]$Operation,

        [Parameter(Mandatory = $true)]
        [string]$AccessToken,

        [Parameter(Mandatory = $true)]
        [string]$ApiBaseUrl,

        [Parameter(Mandatory = $true)]
        [System.IO.FileInfo]$File,

        [Parameter(Mandatory = $true)]
        [string]$OptionsJson
    )

    $form = @{
        file    = $File
        options = $OptionsJson
    }

    $uri = if ($Operation -eq "preview") {
        "$ApiBaseUrl/v1/import/mia/preview"
    }
    else {
        "$ApiBaseUrl/v1/import/mia"
    }

    return Retry-On429 -Script {
        Invoke-RestMethod `
            -Method Post `
            -Uri $uri `
            -Headers @{
                Authorization = "Bearer $AccessToken"
                Accept        = "application/json"
            } `
            -Form $form `
            -TimeoutSec 120 `
            -ErrorAction Stop
    }
}

function Write-JsonResult {
    param(
        [Parameter(Mandatory = $true)]
        [object]$Result
    )

    $Result | ConvertTo-Json -Depth 20 | Write-Host
}

try {
    if ([string]::IsNullOrWhiteSpace($PrivateKeyFile)) {
        throw "PrivateKeyFile is empty. Set it to the path of the EC private key PEM file."
    }

    if (-not (Test-Path -LiteralPath $PrivateKeyFile -PathType Leaf)) {
        throw "Private key file was not found: $PrivateKeyFile"
    }

    if ([string]::IsNullOrWhiteSpace($TenantId)) {
        throw "TenantId is empty."
    }

    if ([string]::IsNullOrWhiteSpace($Kid)) {
        throw "Kid is empty."
    }

    $file = Get-Item -LiteralPath $FilePath -ErrorAction Stop
    if ($file.PSIsContainer) {
        throw "The specified path is a directory, not a file: $FilePath"
    }

    if ($file.Length -eq 0) {
        throw "The import file is empty: $FilePath"
    }

    $maxUploadBytes = 1MB
    if ($file.Length -gt $maxUploadBytes) {
        throw "The import file exceeds the 1 MB API limit."
    }

    Write-Host "-------------------------------------------------------------------"
    Write-Host "MIA Task Import (Automate MFT SaaS)"
    Write-Host "-------------------------------------------------------------------"
    Write-Host "File = $($file.FullName)"
    Write-Host "-------------------------------------------------------------------"

    $passphrase = Read-MiaPassphrase
    $optionsJson = @{
        passphrase     = $passphrase
        targetFolderId = $TargetFolderId
    } | ConvertTo-Json -Compress

    $privateKeyPem = Get-Content -LiteralPath $PrivateKeyFile -Raw -ErrorAction Stop
    $jwt = Create-JWTToken `
        -PrivateKeyPem $privateKeyPem `
        -KeyId $Kid `
        -Tenant $TenantId `
        -ApiBaseUrl $BaseUrl

    $accessToken = Get-AccessToken -JWT $jwt -ApiBaseUrl $BaseUrl

    Write-Host "Preview result:"
    $preview = Invoke-MiaImport `
        -Operation "preview" `
        -AccessToken $accessToken `
        -ApiBaseUrl $BaseUrl `
        -File $file `
        -OptionsJson $optionsJson
    Write-JsonResult -Result $preview

    $previewErrors = @($preview.Errors | Where-Object { $null -ne $_ })
    if ($previewErrors.Count -gt 0) {
        Write-Host ""
        Write-Host "The preview contains $($previewErrors.Count) error(s). The import cannot proceed."
        exit 1
    }

    $confirmation = Read-Host "Preview has no errors. Proceed with import? (y/N)"
    if ($confirmation.Trim().ToLowerInvariant() -notin @("y", "yes")) {
        Write-Host "Import cancelled. No import operation was sent."
        exit 0
    }

    # Refresh the short-lived OAuth token in case the confirmation took a while.
    $jwt = Create-JWTToken `
        -PrivateKeyPem $privateKeyPem `
        -KeyId $Kid `
        -Tenant $TenantId `
        -ApiBaseUrl $BaseUrl
    $accessToken = Get-AccessToken -JWT $jwt -ApiBaseUrl $BaseUrl

    Write-Host "Import result:"
    $importResult = Invoke-MiaImport `
        -Operation "apply" `
        -AccessToken $accessToken `
        -ApiBaseUrl $BaseUrl `
        -File $file `
        -OptionsJson $optionsJson
    Write-JsonResult -Result $importResult
    exit 0
}
catch {
    Write-Error "Error previewing or importing the MIA task: $($_.Exception.Message)"
    exit 1
}

"""
Preview and import a MIA task export using the Automate MFT API.

Dependencies:
    pip install PyJWT requests cryptography
"""

import getpass
import json
import sys
import time
import uuid
from pathlib import Path

import jwt
import requests


# --- Static config -----------------------------------------------------
BASE_URL = "https://api.<env>.mft.progress.com"
PRIVATE_KEY_FILE = ""   # Path to the EC private key PEM file.
TENANT_ID = ""   # Automate MFT tenant ID. You can get it from Plans and Usage page in the Automate MFT UI.
KID = ""   # Key ID of the EC private key. You can get it from API Keys page in the Automate MFT UI.
TARGET_FOLDER_ID = ""   # Automate MFT target folder ID. Leave empty to import into All Tasks. You can get the value by selecting the folder in the Tasks page and get the value of the folderIds query parameter in the URL.
# ----------------------------------------------------------------------

MAX_UPLOAD_BYTES = 10 * 1024 * 1024


def retry_on_429(func, *args, **kwargs):
    """
    Retry a function when the API returns HTTP 429.

    Args:
        func: The function to call.
        *args: Positional arguments passed to the function.
        **kwargs: Keyword arguments passed to the function.

    Returns:
        The function result.
    """
    max_retries = 3
    retry_count = 0

    while retry_count < max_retries:
        try:
            return func(*args, **kwargs)
        except requests.exceptions.HTTPError as error:
            if error.response is None or error.response.status_code != 429:
                raise

            retry_count += 1
            if retry_count < max_retries:
                print(
                    f"Received 429, retrying in 5 seconds... "
                    f"(attempt {retry_count}/{max_retries})"
                )
                time.sleep(5)
            else:
                print("Max retries reached for 429.", file=sys.stderr)
                raise

    raise RuntimeError("Retry loop completed without a result.")


def create_jwt_token(private_key, kid, tenant_id, base_url):
    """
    Create a JWT client assertion for OAuth authentication.

    Args:
        private_key (str): The EC private key content.
        kid (str): The public key ID.
        tenant_id (str): The tenant ID.
        base_url (str): The API base URL.

    Returns:
        str: The signed JWT client assertion.
    """
    now = int(time.time())

    return jwt.encode(
        {
            "aud": f"{base_url}/v1/oauth/token",
            "jti": str(uuid.uuid4()),
            "iat": now,
            "nbf": now,
            "exp": now + 240,
            "iss": tenant_id,
            "sub": tenant_id,
        },
        private_key,
        algorithm="ES256",
        headers={"kid": kid, "typ": "JWT"},
    )


def get_access_token(token, base_url):
    """
    Get an OAuth access token using the JWT client assertion.

    Args:
        token (str): The signed JWT client assertion.
        base_url (str): The API base URL.

    Returns:
        str: The OAuth access token.
    """
    def request_token():
        response = requests.post(
            f"{base_url}/v1/oauth/token",
            headers={"Content-Type": "application/json"},
            json={
                "grant_type": "client_credentials",
                "client_assertion_type": (
                    "urn:ietf:params:oauth:client-assertion-type:jwt-bearer"
                ),
                "client_assertion": token,
            },
            timeout=30,
        )
        response.raise_for_status()
        return response.json()

    token_response = retry_on_429(request_token)
    access_token = token_response.get("access_token")
    if not access_token:
        raise RuntimeError("The OAuth token response did not contain an access_token.")

    return access_token


def invoke_mia_import(operation, access_token, base_url, file_path, options_json):
    """
    Call the MIA preview or apply import endpoint.

    Args:
        operation (str): Either "preview" or "apply".
        access_token (str): The OAuth access token.
        base_url (str): The API base URL.
        file_path (Path): The MIA XML export file.
        options_json (str): JSON for the multipart "options" field.

    Returns:
        dict: The API response.
    """
    if operation == "preview":
        endpoint = "mia/preview"
    elif operation == "apply":
        endpoint = "mia"
    else:
        raise ValueError(f"Unsupported import operation: {operation}")

    def request_import():
        with file_path.open("rb") as import_file:
            response = requests.post(
                f"{base_url}/v1/import/{endpoint}",
                headers={
                    "Authorization": f"bearer {access_token}",
                    "Accept": "application/json",
                },
                files={
                    "file": (
                        file_path.name,
                        import_file,
                        "application/xml",
                    )
                },
                data={"options": options_json},
                timeout=120,
            )
        response.raise_for_status()
        return response.json()

    return retry_on_429(request_import)


def print_json_result(result):
    """Print an API result as formatted JSON."""
    print(json.dumps(result, indent=2))


def validate_file(file_path):
    """
    Validate the MIA XML export file.

    Args:
        file_path (Path): The file to validate.

    Returns:
        Path: The resolved file path.
    """
    if not file_path.exists():
        raise FileNotFoundError(f"Import file was not found: {file_path}")

    if not file_path.is_file():
        raise ValueError(f"The specified path is not a file: {file_path}")

    file_size = file_path.stat().st_size
    if file_size == 0:
        raise ValueError(f"The import file is empty: {file_path}")

    if file_size > MAX_UPLOAD_BYTES:
        raise ValueError("The import file exceeds the 1 MB API limit.")

    return file_path.resolve()


def report_api_error(error):
    """Print an API error response without exposing credentials."""
    response = error.response
    if response is None:
        print(f"API request failed: {error}", file=sys.stderr)
        return

    print(f"API Error: HTTP {response.status_code}", file=sys.stderr)
    try:
        print(json.dumps(response.json(), indent=2), file=sys.stderr)
    except ValueError:
        print(response.text, file=sys.stderr)


def call_apis(file_path):
    """
    Preview and optionally apply a MIA task import.

    Args:
        file_path (Path): The MIA XML export file.
    """
    if not PRIVATE_KEY_FILE:
        raise ValueError("PRIVATE_KEY_FILE is empty.")

    private_key_path = Path(PRIVATE_KEY_FILE)
    if not private_key_path.is_file():
        raise FileNotFoundError(
            f"Private key file was not found: {private_key_path}"
        )

    if not TENANT_ID:
        raise ValueError("TENANT_ID is empty.")

    if not KID:
        raise ValueError("KID is empty.")

    file_path = validate_file(file_path)

    print("-------------------------------------------------------------------")
    print("MIA Task Import (Automate MFT SaaS)")
    print("-------------------------------------------------------------------")
    print(f"File = {file_path}")
    print("-------------------------------------------------------------------")

    passphrase = getpass.getpass(
        "MIA export passphrase (press Enter if none): "
    )
    options_json = json.dumps(
        {
            "passphrase": passphrase,
            "targetFolderId": TARGET_FOLDER_ID,
        },
        separators=(",", ":"),
    )

    private_key = private_key_path.read_text(encoding="utf-8")
    token = create_jwt_token(private_key, KID, TENANT_ID, BASE_URL)
    access_token = get_access_token(token, BASE_URL)

    print("Preview result:")
    preview = invoke_mia_import(
        "preview",
        access_token,
        BASE_URL,
        file_path,
        options_json,
    )
    print_json_result(preview)

    preview_errors = preview.get("errors") or []
    if preview_errors:
        print()
        print(
            f"The preview contains {len(preview_errors)} error(s). "
            "The import cannot proceed."
        )
        return

    confirmation = input(
        "Preview has no errors. Proceed with import? (y/N): "
    ).strip().lower()
    if confirmation not in {"y", "yes"}:
        print("Import cancelled. No import operation was sent.")
        return

    # Refresh the short-lived OAuth token in case confirmation took a while.
    token = create_jwt_token(private_key, KID, TENANT_ID, BASE_URL)
    access_token = get_access_token(token, BASE_URL)

    print("Import result:")
    import_result = invoke_mia_import(
        "apply",
        access_token,
        BASE_URL,
        file_path,
        options_json,
    )
    print_json_result(import_result)


if __name__ == "__main__":
    if len(sys.argv) != 2:
        print(
            "Usage: python runimport.py <path_to_mia_xml_file>",
            file=sys.stderr,
        )
        sys.exit(1)

    try:
        call_apis(Path(sys.argv[1]))
    except requests.exceptions.RequestException as error:
        report_api_error(error)
        sys.exit(1)
    except Exception as error:
        print(f"Error previewing or importing the MIA task: {error}", file=sys.stderr)
        sys.exit(1)