> ## Documentation Index
> Fetch the complete documentation index at: https://docs.corgea.app/llms.txt
> Use this file to discover all available pages before exploring further.

# Carga de escaneos por fragmentos

> Sube escaneos BLAST grandes por fragmentos con la API Start Scan

# Carga de escaneos por fragmentos

La API Start Scan sube bases de código grandes en varias solicitudes. El escaneo BLAST debe estar habilitado para tu empresa y solo puedes subir un archivo de origen.

Esta guía describe la secuencia de carga. Los campos de solicitud y los códigos de estado están en las páginas OpenAPI de `POST /start-scan`, `PATCH /start-scan/{transfer_id}/` y `HEAD /start-scan/{transfer_id}/`.

## Secuencia de carga

1. **Inicia la transferencia** con `POST /start-scan`. Envía `scan_type=blast` como campo de formulario y una parte `files` con el nombre del archivo y un cuerpo vacío. El nombre del archivo debe usar una de estas extensiones: `.zip`, `.tar`, `.json`, `.fpr`, `.sarif` o `.xml`. La respuesta incluye un `transfer_id`.
2. **Lee el desplazamiento actual** con `HEAD /start-scan/{transfer_id}/`. La cabecera de respuesta `Upload-Offset` indica el siguiente byte que debes enviar. Una transferencia desconocida o vacía devuelve `0`.
3. **Sube cada fragmento** con `PATCH /start-scan/{transfer_id}/`. Envía los bytes siguientes como `chunk_data` e incluye estas cabeceras de solicitud:
   * `Upload-Offset`: desplazamiento actual en bytes
   * `Upload-Length`: tamaño total del archivo en bytes
   * `Upload-Name`: nombre del archivo
4. **Termina cuando el último PATCH devuelve un `scan_id`.** Las respuestas intermedias incluyen `Upload-Offset` (siguiente byte) y no incluyen `scan_id`. Cuando el último fragmento completa el archivo, el cuerpo JSON incluye `scan_id` y `project_id`. Esa respuesta final no establece `Upload-Offset`. Usa el `scan_id` con las APIs de escaneos y de problemas. HEAD no devuelve `scan_id`.

También puedes enviar `project_name`, `branch`, `repo_url`, `sha`, `files_to_scan`, `dirty`, `scan_configs` y `target_policies` en cada solicitud PATCH. Omite `partial_scan` para un escaneo completo. Envía `partial_scan=true` solo para un escaneo parcial.

## Autenticación

Incluye tu token de API en cada solicitud:

```http theme={null}
CORGEA-TOKEN: your_api_token_here
```

## Metadatos del escaneo

`metadata` es una cadena de objeto JSON opcional, por ejemplo `{"pipeline_url": "https://ci.example/run/123"}`. Corgea la valida en cada solicitud de fragmento y la adjunta al escaneo cuando termina la carga, así que envía el mismo valor con cada fragmento. El objeto no debe superar los 16.384 bytes.

## Ejemplo en Python

Este script inicia una carga BLAST o la reanuda con `--transfer_id`. Sigue `Upload-Offset` en los fragmentos intermedios. El último PATCH devuelve `scan_id` en lugar de `Upload-Offset`.

```python theme={null}
import argparse
import os

import requests

API_BASE_URL = "https://www.corgea.app/api/v1/start-scan"
CHUNK_SIZE = 5 * 1024 * 1024  # 5 MB example; the CLI uses 50 MB. Any size is valid.
TOKEN_HEADERS = {"CORGEA-TOKEN": "<YOUR_TOKEN>"}


def transfer_url(transfer_id):
    """HEAD and PATCH continue URLs require a trailing slash."""
    return f"{API_BASE_URL}/{transfer_id}/"


def require_upload_offset(headers):
    """Return the server's next-byte offset. Fail if the header is missing."""
    value = headers.get("Upload-Offset")
    if value is None:
        raise RuntimeError("Response is missing the required Upload-Offset header.")
    return int(value)


def initiate_upload(file_path):
    """Start the transfer with the file name and an empty first part."""
    file_name = os.path.basename(file_path)
    response = requests.post(
        API_BASE_URL,
        data={"scan_type": "blast"},
        files={"files": (file_name, b"")},
        headers=TOKEN_HEADERS,
    )
    response.raise_for_status()
    transfer_id = response.json().get("transfer_id")
    if not transfer_id:
        raise RuntimeError("Start-scan response is missing transfer_id.")
    print(f"Upload initiated. Transfer ID: {transfer_id}")
    return transfer_id


def check_upload_status(transfer_id):
    """Return the next byte offset from the Upload-Offset header."""
    response = requests.head(
        transfer_url(transfer_id),
        headers=TOKEN_HEADERS,
    )
    response.raise_for_status()
    return require_upload_offset(response.headers)


def upload_chunk(
    file_path,
    transfer_id,
    chunk_offset,
    chunk_data,
    project_name,
    branch,
    repo_url,
    sha,
    partial_scan,
    files_to_scan,
    metadata="",
):
    headers = {
        "Upload-Offset": str(chunk_offset),
        "Upload-Length": str(os.path.getsize(file_path)),
        "Upload-Name": os.path.basename(file_path),
        **TOKEN_HEADERS,
    }
    form_data = {}
    if project_name:
        form_data["project_name"] = project_name
    if branch:
        form_data["branch"] = branch
    if repo_url:
        form_data["repo_url"] = repo_url
    if sha:
        form_data["sha"] = sha
    if partial_scan:
        form_data["partial_scan"] = "true"
    if files_to_scan:
        form_data["files_to_scan"] = files_to_scan
    if metadata:
        form_data["metadata"] = metadata
    response = requests.patch(
        transfer_url(transfer_id),
        headers=headers,
        files={"chunk_data": ("chunk", chunk_data, "application/octet-stream")},
        data=form_data,
    )
    response.raise_for_status()
    return response.json(), response.headers


def upload_file_in_chunks(
    file_path,
    project_name,
    branch,
    repo_url,
    sha,
    partial_scan=False,
    files_to_scan="",
    transfer_id=None,
    metadata="",
):
    file_size = os.path.getsize(file_path)
    if transfer_id:
        print(f"Resuming transfer {transfer_id}")
    else:
        transfer_id = initiate_upload(file_path)
    offset = check_upload_status(transfer_id)
    print(f"Uploading {file_path} ({file_size} bytes) from offset {offset}")

    scan_id = None
    with open(file_path, "rb") as handle:
        while offset < file_size:
            handle.seek(offset)
            chunk_data = handle.read(CHUNK_SIZE)
            if not chunk_data:
                break
            payload, response_headers = upload_chunk(
                file_path,
                transfer_id,
                offset,
                chunk_data,
                project_name,
                branch,
                repo_url,
                sha,
                partial_scan,
                files_to_scan,
                metadata,
            )
            scan_id = payload.get("scan_id")
            if scan_id:
                break
            message = payload.get("message") or ""
            if (
                payload.get("status") == "ok"
                and "already in progress" in message.lower()
            ):
                print(message)
                return
            next_offset = require_upload_offset(response_headers)
            if next_offset <= offset:
                raise RuntimeError(
                    f"Upload-Offset did not advance (was {offset}, now {next_offset})."
                )
            if next_offset > file_size:
                raise RuntimeError(
                    f"Upload-Offset {next_offset} exceeds file size {file_size}."
                )
            offset = next_offset
            print(f"Uploaded to server offset {offset} / {file_size} bytes")

    if not scan_id:
        raise RuntimeError(
            "Upload finished without a scan_id. HEAD /start-scan/{transfer_id}/ does not return the scan_id after a completed transfer; look it up with the scans API."
        )
    print(f"Upload complete. Scan ID: {scan_id}")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Upload a BLAST archive in chunks.")
    parser.add_argument("file", help="Path to the archive to upload")
    parser.add_argument("--branch", required=False)
    parser.add_argument("--repo_url", required=False)
    parser.add_argument("--sha", required=False)
    parser.add_argument("--project_name", required=False)
    parser.add_argument("--partial_scan", action="store_true")
    parser.add_argument(
        "--files_to_scan",
        required=False,
        default="",
        help="Comma-separated files for a partial scan",
    )
    parser.add_argument(
        "--transfer_id",
        required=False,
        help="Resume an existing transfer instead of starting a new one",
    )
    parser.add_argument(
        "--metadata",
        required=False,
        default="",
        help='JSON object string sent with every chunk, e.g. {"pipeline_url":"https://ci.example/run/123"}',
    )
    args = parser.parse_args()

    if not os.path.isfile(args.file):
        raise SystemExit(f"File '{args.file}' does not exist.")

    upload_file_in_chunks(
        args.file,
        args.project_name,
        args.branch,
        args.repo_url,
        args.sha,
        args.partial_scan,
        args.files_to_scan,
        args.transfer_id,
        args.metadata,
    )
```

Ejemplo:

```bash theme={null}
python upload_scan_chunks.py \
  --branch main \
  --repo_url https://github.com/example/repo \
  --sha 21jio112j3 \
  --project_name projectX \
  /path/to/source.zip
```

Para reanudar una carga interrumpida, pasa el ID de transferencia existente:

```bash theme={null}
python upload_scan_chunks.py \
  --transfer_id c9b0a8c7-f9b4-4c10-9d58-cd4c7e1c9c52 \
  --branch main \
  --repo_url https://github.com/example/repo \
  --sha 21jio112j3 \
  --project_name projectX \
  /path/to/source.zip
```

Para un escaneo parcial, añade `--partial_scan` y `--files_to_scan "vuln.py,test.py"`.

Para adjuntar metadatos del escaneo, añade `--metadata '{"pipeline_url":"https://ci.example/run/123"}'`. Se envía el mismo valor con cada fragmento.