Securely Importing and Exporting Files in Air-Gapped and Network-Segmented Environments

import os
import requests

BASE_URL = os.getenv("INNORIX_BASE_URL", "https://app.innorix.com").rstrip("/")
TOKEN = os.environ["INNORIX_ACCESS_TOKEN"]
WORKSPACE_ID = os.getenv("INNORIX_WORKSPACE_ID")   # optional; falls back to the current workspace

STATUS_COMPLETE = 2
TERMINAL = {2, 4, 5, 9, 99}          # complete / error / cancelled / partial / failed
NOT_SUCCEEDED = {4, 5, 9, 99}


def api(method, path, body=None, params=None):
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {TOKEN}",
    }

    if WORKSPACE_ID:
        headers["x-workspace-id"] = WORKSPACE_ID

    response = requests.request(
        method, BASE_URL + path,
        headers=headers, json=body, params=params, timeout=30,
    )

    payload = response.json() if response.content else {}

    if not response.ok:
        raise RuntimeError(payload.get("message") or f"HTTP {response.status_code}")

    return payload.get("data")


def is_terminal(detail):
    return detail.get("isTerminal", detail.get("status") in TERMINAL)

def client_for(workspace_id):
    def call(method, path, body=None, params=None):
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {TOKEN}",
            "x-workspace-id": workspace_id,
        }

        response = requests.request(method, BASE_URL + path, headers=headers,
                                    json=body, params=params, timeout=30)

        payload = response.json() if response.content else {}

        if response.status_code == 403:
            raise PermissionError(f"{workspace_id}: {path}")

        if not response.ok:
            raise RuntimeError(payload.get("message"))

        return payload.get("data")

    return call


external = client_for(EXTERNAL_WORKSPACE)   # external network
internal = client_for(INTERNAL_WORKSPACE)   # internal network

import uuid


def create_request(user_id, direction, files, purpose):
    request_id = str(uuid.uuid4())

    db.insert("transfer_requests", {
        "id": request_id,
        "userId": user_id,
        "direction": direction,          # "import" or "export"
        "files": files,
        "purpose": purpose,
        "state": "pending",
    })

    return request_id

def validate_request(call, source_id, target_id, files, target_path):
    # sourceItems reads filePath, not path
    result = call("POST", "/api/transfers/validate-path", {
        "sourceId": source_id,
        "targetId": target_id,
        "sourceItems": [{"filePath": p} for p in files],
        "targetPath": target_path,
    }) or {}

    if result.get("invalidSourcePaths"):
        raise ValueError(f"missing source paths: {result['invalidSourcePaths']}")

    if result.get("validTargetPath") is False:
        raise ValueError(f"target path not found: {target_path}")

    return result

def on_approval(request_id, approved, approver):
    request = db.get("transfer_requests", request_id)

    db.update("transfer_requests", request_id, {
        "state": "approved" if approved else "rejected",
        "approver": approver,
    })

    if not approved:
        return None

    return send_for_scan(request)


def ensure_approved(request):
    if request["state"] not in ("approved", "validated"):
        raise PermissionError(f"request not approved: {request['id']}")

def send_for_scan(request, scan_device, scan_path):
    ensure_approved(request)

    transfer = external("POST", "/api/transfers/manual", {
        "sourceDevice": request["sourceDevice"],
        "targetDevice": scan_device,
        "targetPath": f"{scan_path}/{request['id']}",
        "sourceItem": [{"path": p, "isDir": False} for p in request["files"]],
        "sendAllFolder": False,
        "checkIntegrity": True,
        "transferOptions": {"target-action": "numbering"},
    })

    db.update("transfer_requests", request["id"], {
        "scanMonitorId": transfer["monitorId"],
        "state": "scanning",
    })

    return transfer["monitorId"]

def verify(monitor_id, timeout=1800, interval=10):
    api("POST", f"/api/transfers/{monitor_id}/verification", {})

    deadline = time.time() + timeout

    while time.time() < deadline:
        result = api("GET", f"/api/transfers/{monitor_id}/verification") or {}

        if result.get("verified"):
            return result

        time.sleep(interval)

    raise TimeoutError(monitor_id)

def execute(request, scan_result):
    if not scan_result.get("passed"):
        db.update("transfer_requests", request["id"], {"state": "blocked"})
        return None

    verification = verify(request["scanMonitorId"])

    if not verification.get("checksumMatched"):
        db.update("transfer_requests", request["id"], {"state": "corrupted"})
        return None

    transfer = internal("POST", "/api/transfers/manual", {
        "sourceDevice": request["scanDevice"],
        "targetDevice": request["targetDevice"],
        "targetPath": request["targetPath"],
        "sourcePaths": [f"{request['scanPath']}/{request['id']}"],
        "sendAllFolder": True,
        "checkIntegrity": True,
        "transferOptions": {"target-action": "numbering"},
    })

    db.update("transfer_requests", request["id"], {
        "monitorId": transfer["monitorId"],
        "state": "transferring",
    })

    return transfer["monitorId"]

def wait(monitor_id, timeout=3600, interval=3):
    deadline = time.time() + timeout

    while time.time() < deadline:
        detail = api("GET", f"/api/transfers/{monitor_id}")

        if is_terminal(detail):
            return detail

        time.sleep(interval)

    raise TimeoutError(monitor_id)


def failed_files(monitor_id):
    result = api("GET", f"/api/transfers/{monitor_id}/files", params={
        "state": "any", "size": 500,
    }) or {}

    return [r for r in (result.get("children") or [])
            if r.get("status") in NOT_SUCCEEDED]


def retry_failed(monitor_id):
    rows = failed_files(monitor_id)

    if not rows:
        return 0

    api("POST", f"/api/transfers/{monitor_id}/retry", {
        "filesRetry": [
            {"filePath": r["sourceFilePath"], "isDir": bool(r.get("isFolder"))}
            for r in rows
        ]
    })

    return len(rows)

def finalize(request_id):
    request = db.get("transfer_requests", request_id)
    detail = internal("GET", f"/api/transfers/{request['monitorId']}")

    if not is_terminal(detail):
        return None

    succeeded = detail["status"] == STATUS_COMPLETE

    db.update("transfer_requests", request_id, {
        "state": "done" if succeeded else "failed",
        "finalStatus": detail["status"],
    })

    return succeeded

from datetime import datetime, timedelta, timezone


def paginate(call, path, params=None, limit=200, max_pages=50):
    query = dict(params or {})
    query["limit"] = limit
    cursor = None

    for _ in range(max_pages):
        if cursor:
            query["cursor"] = cursor

        result = call("GET", path, params=query) or {}

        for record in result.get("data") or []:
            yield record

        pagination = result.get("pagination") or {}

        if not pagination.get("hasMore"):
            return

        cursor = pagination.get("nextCursor")

        if not cursor:
            return


end = datetime.now(timezone.utc)
fmt = "%Y-%m-%dT%H:%M:%SZ"

for row in paginate(internal, "/api/transfer-history", params={
    "startDate": (end - timedelta(days=30)).strftime(fmt),
    "endDate": end.strftime(fmt),
}):
    print(row.get("startDate"), row.get("statusName"),
          row.get("sourceDeviceName"), "->", row.get("targetDeviceName"))

params = {
    "periodDays": 30,
    "page": 1,
    "size": 10000,
    "filter": "[]",              # the server parses this as a JSON string, so send an empty array
    "sort": "startDate:desc",
}

def review(request_id):
    request = db.get("transfer_requests", request_id)
    state = request["state"]

    if state == "rejected":
        return "rejected - notify the requester with the reason"

    if state == "blocked":
        return "scan blocked - check the files and request again"

    if state == "corrupted":
        return "integrity mismatch - check the source"

    if state == "transferring":
        return f"retried {retry_failed(request['monitorId'])} failed files"

    return f"current state: {state}"