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

# Chargement d’un scan par blocs

> Charger de grands scans BLAST par blocs avec l’API Start Scan

# Chargement d’un scan par blocs

L’API Start Scan envoie les grandes bases de code en plusieurs requêtes. Le scan BLAST doit être activé pour votre entreprise, et vous chargez une seule archive source.

Ce guide décrit la séquence de chargement. Les champs de requête et les codes de statut se trouvent sur les pages OpenAPI de `POST /start-scan`, `PATCH /start-scan/{transfer_id}/` et `HEAD /start-scan/{transfer_id}/`.

## Séquence de chargement

1. **Démarrer le transfert** avec `POST /start-scan`. Envoyez `scan_type=blast` comme champ de formulaire et une partie `files` qui contient le nom de l’archive et un corps vide. Le nom de l’archive doit utiliser l’une des extensions `.zip`, `.tar`, `.json`, `.fpr`, `.sarif` ou `.xml`. La réponse inclut un `transfer_id`.
2. **Lire l’offset actuel** avec `HEAD /start-scan/{transfer_id}/`. L’en-tête de réponse `Upload-Offset` indique le prochain octet à envoyer. Un transfert inconnu ou vide renvoie `0`.
3. **Charger chaque bloc** avec `PATCH /start-scan/{transfer_id}/`. Envoyez les octets suivants dans `chunk_data` et incluez ces en-têtes de requête :
   * `Upload-Offset` : offset actuel en octets
   * `Upload-Length` : taille totale du fichier en octets
   * `Upload-Name` : nom de l’archive
4. **Terminer lorsque le dernier PATCH renvoie un `scan_id`.** Les réponses intermédiaires incluent `Upload-Offset` (prochain octet) et n’incluent pas de `scan_id`. Lorsque le dernier bloc complète l’archive, le corps JSON contient `scan_id` et `project_id`. Cette réponse finale ne définit pas `Upload-Offset`. Utilisez `scan_id` avec les API de scans et de problèmes. HEAD ne renvoie pas de `scan_id`.

Vous pouvez aussi envoyer `project_name`, `branch`, `repo_url`, `sha`, `files_to_scan`, `dirty`, `scan_configs` et `target_policies` à chaque requête PATCH. Omettez `partial_scan` pour un scan complet. Envoyez `partial_scan=true` uniquement pour un scan partiel.

## Authentification

Incluez votre jeton d’API dans chaque requête :

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

## Métadonnées de scan

`metadata` est une chaîne JSON d’objet facultative, par exemple `{"pipeline_url": "https://ci.example/run/123"}`. Corgea la valide à chaque requête de bloc et l’associe au scan une fois le chargement terminé : envoyez donc la même valeur avec chaque bloc. L’objet ne doit pas dépasser 16 384 octets.

## Exemple Python

Ce script démarre un chargement BLAST ou le reprend avec `--transfer_id`. Il suit `Upload-Offset` sur les blocs intermédiaires. Le dernier PATCH renvoie `scan_id` au lieu 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,
    )
```

Exemple :

```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
```

Pour reprendre un chargement interrompu, passez l’ID de transfert existant :

```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
```

Pour un scan partiel, ajoutez `--partial_scan` et `--files_to_scan "vuln.py,test.py"`.

Pour joindre des métadonnées de scan, ajoutez `--metadata '{"pipeline_url":"https://ci.example/run/123"}'`. La même valeur est envoyée avec chaque bloc.