diff --git a/docs/external-sources/easm/how-to/010_easm-source.md b/docs/external-sources/easm/how-to/010_easm-source.md
new file mode 100644
index 0000000..342c56e
--- /dev/null
+++ b/docs/external-sources/easm/how-to/010_easm-source.md
@@ -0,0 +1,96 @@
+# External Attack Surface Management (EASM) Integration
+
+The EASM Integration of `ibdm` connects to the EASM backend via its REST API and reads a
+customer's latest completed scan run. It supports pull-based sync only.
+
+## Commands
+
+### Sync
+
+```sh
+ibdm sync easm --mapping-file <path to mapping file or folder>
+```
+
+Performs a one-off synchronisation: fetches the customer's latest completed run from the EASM
+`/data` endpoint as a single cursor-paginated list, emits one item per record routed by the
+record's own `type` field, and exits.
+
+## Configuration
+
+All configuration is read from environment variables.
+
+### Environment Variables
+
+| Env Variable | Required | Default | Description |
+| --- | --- | --- | --- |
+| `EASM_BASE_URL` | Yes | _(empty)_ | Base URL of the EASM backend (e.g. `https://easm.example.com`). |
+| `EASM_CUSTOMER` | Yes | _(empty)_ | Customer identifier. Scopes the request to a single customer via the `X-Customer` header — it selects whose scan results to read. |
+| `EASM_TOKEN` | No | _(empty)_ | Bearer token authenticating the caller. When set, it is sent as `Authorization: Bearer <token>`. |
+| `EASM_DATA_PATH` | No | `/data` | Path of the read endpoint appended to `EASM_BASE_URL`. |
+| `EASM_HTTP_TIMEOUT` | No | `30s` | Timeout for each HTTP request, parsed as a Go `time.Duration`. |
+
+## Supported Data Types
+
+| Type | Sync |
+| --- | --- |
+| `domain` | ✅ |
+| `endpoint` | ✅ |
+| `host` | ✅ |
+| `ip` | ✅ |
+| `vulnerability` | ✅ |
+
+The endpoint tags each record with one of these types; records with a missing or empty `type`
+are skipped. Each emitted item carries the record's fields unchanged, and the pipeline can restrict
+a run to a subset of types via the mapping files — only the requested types are emitted.
+
+### `domain`
+
+One entry per discovered domain, including its DNS, WHOIS, zone-transfer, Azure, and
+misconfiguration data.
+
+### `endpoint`
+
+One entry per discovered endpoint.
+
+### `host`
+
+One entry per discovered host.
+
+### `ip`
+
+One entry per discovered IP address.
+
+### `vulnerability`
+
+One entry per discovered vulnerability.
+
+## Authentication
+
+The source scopes every request to a single customer with the `X-Customer` header, taken from
+`EASM_CUSTOMER` — this is always sent.
+
+When `EASM_TOKEN` is set, the source authenticates the caller with an
+`Authorization: Bearer <token>` header.
+
+## Example Mapping Files
+
+Example mapping files are provided in the `docs/external-sources/easm/mappings/` directory:
+
+- `domains.yaml` — maps domain records to Catalog items.
+- `endpoints.yaml` — maps endpoint records to Catalog items.
+- `hosts.yaml` — maps host records to Catalog items.
+- `ips.yaml` — maps IP records to Catalog items.
+- `vulnerabilities.yaml` — maps vulnerability records to Catalog items.
+
+These files can be used as a starting point for your own mapping configuration. Pass the file
+or the folder to the `--mapping-file` flag:
+
+```sh
+ibdm sync easm --mapping-file docs/external-sources/easm/mappings/
+```
+
+For local development and debugging, add the `--local-output` flag to send results to stdout:
+
+```sh
+ibdm sync easm --mapping-file docs/external-sources/easm/mappings/ --local-output
+```
diff --git a/docs/external-sources/easm/itds/domains.easm.unguess.io.yaml b/docs/external-sources/easm/itds/domains.easm.unguess.io.yaml
new file mode 100644
index 0000000..ca03339
--- /dev/null
+++ b/docs/external-sources/easm/itds/domains.easm.unguess.io.yaml
@@ -0,0 +1,102 @@
+# yaml-language-server: $schema=https://cdn.mia-platform.eu/catalog/v1alpha1/mia-platform.eu.v1alpha1.ItemTypeDefinition.json
+---
+apiVersion: easm.unguess.io/v1
+kind: ItemTypeDefinition
+metadata:
+  name: domains.easm.unguess.io
+spec:
+  group: easm.unguess.io
+  names:
+    kind: Domain
+    plural: domains
+    singular: domain
+    displayPlural: Domains
+    displaySingular: Domain
+  scope: Organization
+  versions:
+    - deprecated: false
+      name: v1
+      schema:
+        openAPIV31Schema:
+          type: object
+          properties:
+            spec:
+              type: object
+              properties:
+                domain:
+                  description: The registered domain name.
+                  type: string
+                  title: Domain
+                dns:
+                  description: DNS records found for the domain.
+                  type: object
+                  title: DNS
+                  properties:
+                    dmarc:
+                      type: string
+                      title: DMARC
+                    spf:
+                      type: string
+                      title: SPF
+                    mx:
+                      type: array
+                      title: MX
+                      items:
+                        type: string
+                    ns:
+                      type: array
+                      title: NS
+                      items:
+                        type: string
+                whois:
+                  description: The raw WHOIS record for the domain (irregular structure, stored as-is).
+                  type: object
+                  title: WHOIS
+                  additionalProperties: true
+                zonetransfer:
+                  description: AXFR zone-transfer lines ("name\ttype\trdata"); absent when blocked.
+                  type: array
+                  title: Zone Transfer
+                  items:
+                    type: string
+                azure:
+                  description: Azure tenant attribution for the domain (absent when skipped or nothing found).
+                  type: object
+                  title: Azure
+                  properties:
+                    tenant_id:
+                      type: string
+                      title: Tenant ID
+                    namespace_type:
+                      type: string
+                      title: Namespace Type
+                    brand:
+                      type: string
+                      title: Brand
+                    federation_url:
+                      type: string
+                      title: Federation URL
+                    m365:
+                      type: boolean
+                      title: Microsoft 365
+                    admin_consent_exposed:
+                      type: boolean
+                      title: Admin Consent Exposed
+                misconfig:
+                  description: Cloud service misconfigurations discovered for the domain.
+                  type: array
+                  title: Misconfigurations
+                  items:
+                    type: object
+                    properties:
+                      service:
+                        type: string
+                        title: Service
+                      url:
+                        type: string
+                        title: URL
+                      status:
+                        type: integer
+                        title: Status
+      served: true
+      storage: true
diff --git a/docs/external-sources/easm/itds/endpoints.easm.unguess.io.yaml b/docs/external-sources/easm/itds/endpoints.easm.unguess.io.yaml
new file mode 100644
index 0000000..410bc94
--- /dev/null
+++ b/docs/external-sources/easm/itds/endpoints.easm.unguess.io.yaml
@@ -0,0 +1,67 @@
+# yaml-language-server: $schema=https://cdn.mia-platform.eu/catalog/v1alpha1/mia-platform.eu.v1alpha1.ItemTypeDefinition.json
+---
+apiVersion: easm.unguess.io/v1
+kind: ItemTypeDefinition
+metadata:
+  name: endpoints.easm.unguess.io
+spec:
+  group: easm.unguess.io
+  names:
+    kind: Endpoint
+    plural: endpoints
+    singular: endpoint
+    displayPlural: Endpoints
+    displaySingular: Endpoint
+  scope: Organization
+  versions:
+    - deprecated: false
+      name: v1
+      schema:
+        openAPIV31Schema:
+          type: object
+          properties:
+            spec:
+              type: object
+              properties:
+                host:
+                  description: The host the endpoint belongs to.
+                  type: string
+                  title: Host
+                port:
+                  description: The port the endpoint listens on.
+                  type: integer
+                  title: Port
+                http:
+                  description: HTTP response details for the endpoint (absent when no HTTP answer).
+                  type: object
+                  title: HTTP
+                  properties:
+                    url:
+                      type: string
+                      title: URL
+                    scheme:
+                      type: string
+                      title: Scheme
+                    status_codes:
+                      type: array
+                      title: Status Codes
+                      items:
+                        type: integer
+                    content_length:
+                      type: integer
+                      title: Content Length
+                    title:
+                      type: string
+                      title: Title
+                    technologies:
+                      type: array
+                      title: Technologies
+                      items:
+                        type: string
+                tls:
+                  description: TLS report for the endpoint (success shape or an error; stored as-is).
+                  type: object
+                  title: TLS
+                  additionalProperties: true
+      served: true
+      storage: true
diff --git a/docs/external-sources/easm/itds/hosts.easm.unguess.io.yaml b/docs/external-sources/easm/itds/hosts.easm.unguess.io.yaml
new file mode 100644
index 0000000..0a53fb5
--- /dev/null
+++ b/docs/external-sources/easm/itds/hosts.easm.unguess.io.yaml
@@ -0,0 +1,53 @@
+# yaml-language-server: $schema=https://cdn.mia-platform.eu/catalog/v1alpha1/mia-platform.eu.v1alpha1.ItemTypeDefinition.json
+---
+apiVersion: easm.unguess.io/v1
+kind: ItemTypeDefinition
+metadata:
+  name: hosts.easm.unguess.io
+spec:
+  group: easm.unguess.io
+  names:
+    kind: Host
+    plural: hosts
+    singular: host
+    displayPlural: Hosts
+    displaySingular: Host
+  scope: Organization
+  versions:
+    - deprecated: false
+      name: v1
+      schema:
+        openAPIV31Schema:
+          type: object
+          properties:
+            spec:
+              type: object
+              properties:
+                fqdn:
+                  description: The fully qualified domain name of the host.
+                  type: string
+                  title: FQDN
+                status_code:
+                  description: The DNS response status observed for the host.
+                  type: string
+                  title: Status Code
+                a:
+                  description: The IPv4 (A) records the host resolves to.
+                  type: array
+                  title: A Records
+                  items:
+                    type: string
+                aaaa:
+                  description: The IPv6 (AAAA) records the host resolves to.
+                  type: array
+                  title: AAAA Records
+                  items:
+                    type: string
+                cname:
+                  description: The CNAME records for the host.
+                  type: array
+                  title: CNAME Records
+                  items:
+                    type: string
+      served: true
+      storage: true
diff --git a/docs/external-sources/easm/itds/ips.easm.unguess.io.yaml b/docs/external-sources/easm/itds/ips.easm.unguess.io.yaml
new file mode 100644
index 0000000..a209c12
--- /dev/null
+++ b/docs/external-sources/easm/itds/ips.easm.unguess.io.yaml
@@ -0,0 +1,61 @@
+# yaml-language-server: $schema=https://cdn.mia-platform.eu/catalog/v1alpha1/mia-platform.eu.v1alpha1.ItemTypeDefinition.json
+---
+apiVersion: easm.unguess.io/v1
+kind: ItemTypeDefinition
+metadata:
+  name: ips.easm.unguess.io
+spec:
+  group: easm.unguess.io
+  names:
+    kind: Ip
+    plural: ips
+    singular: ip
+    displayPlural: IPs
+    displaySingular: IP
+  scope: Organization
+  versions:
+    - deprecated: false
+      name: v1
+      schema:
+        openAPIV31Schema:
+          type: object
+          properties:
+            spec:
+              type: object
+              properties:
+                ip:
+                  description: The IP address.
+                  type: string
+                  title: IP
+                ptr:
+                  description: The PTR (reverse DNS) record for the IP.
+                  type: string
+                  title: PTR
+                ports:
+                  description: The open ports discovered on the IP.
+                  type: array
+                  title: Ports
+                  items:
+                    type: integer
+                geo:
+                  description: Geographic and network attribution for the IP.
+                  type: object
+                  title: Geo
+                  properties:
+                    provider:
+                      type: string
+                      title: Provider
+                    org:
+                      type: string
+                      title: Organization
+                    asn:
+                      type: string
+                      title: ASN
+                    region:
+                      type: string
+                      title: Region
+                    country:
+                      type: string
+                      title: Country
+      served: true
+      storage: true
diff --git a/docs/external-sources/easm/itds/vulnerabilities.easm.unguess.io.yaml b/docs/external-sources/easm/itds/vulnerabilities.easm.unguess.io.yaml
new file mode 100644
index 0000000..7bb1cba
--- /dev/null
+++ b/docs/external-sources/easm/itds/vulnerabilities.easm.unguess.io.yaml
@@ -0,0 +1,86 @@
+# yaml-language-server: $schema=https://cdn.mia-platform.eu/catalog/v1alpha1/mia-platform.eu.v1alpha1.ItemTypeDefinition.json
+---
+apiVersion: easm.unguess.io/v1
+kind: ItemTypeDefinition
+metadata:
+  name: vulnerabilities.easm.unguess.io
+spec:
+  group: easm.unguess.io
+  names:
+    kind: Vulnerability
+    plural: vulnerabilities
+    singular: vulnerability
+    displayPlural: Vulnerabilities
+    displaySingular: Vulnerability
+  scope: Organization
+  versions:
+    - deprecated: false
+      name: v1
+      schema:
+        openAPIV31Schema:
+          type: object
+          properties:
+            spec:
+              type: object
+              properties:
+                template_id:
+                  description: Identifier of the detection rule that produced the finding.
+                  type: string
+                  title: Template ID
+                name:
+                  description: The name of the finding.
+                  type: string
+                  title: Name
+                severity:
+                  description: The severity of the finding (info, low, medium, high, critical).
+                  type: string
+                  title: Severity
+                target:
+                  description: The target the finding was matched on.
+                  type: string
+                  title: Target
+                description:
+                  description: A human-readable description of the finding.
+                  type: string
+                  title: Description
+                tags:
+                  description: The tags associated with the finding.
+                  type: array
+                  title: Tags
+                  items:
+                    type: string
+                extracted_results:
+                  description: Values extracted by the detection rule (when any).
+                  type: array
+                  title: Extracted Results
+                  items:
+                    type: string
+                matcher_name:
+                  description: Name of the matcher that triggered the finding.
+                  type: string
+                  title: Matcher Name
+                scheme:
+                  description: The scheme the finding was observed on.
+                  type: string
+                  title: Scheme
+                meta:
+                  description: Additional rule-specific metadata for the finding (free-form).
+                  type: object
+                  title: Meta
+                  additionalProperties: true
+                owasp:
+                  description: The OWASP Top 10 category associated with the finding.
+                  type: object
+                  title: OWASP
+                  properties:
+                    code:
+                      type: string
+                      title: Code
+                    name:
+                      type: string
+                      title: Name
+                    url:
+                      type: string
+                      title: URL
+      served: true
+      storage: true
diff --git a/docs/external-sources/easm/mappings/domains.yaml b/docs/external-sources/easm/mappings/domains.yaml
new file mode 100644
index 0000000..cd35361
--- /dev/null
+++ b/docs/external-sources/easm/mappings/domains.yaml
@@ -0,0 +1,15 @@
+apiVersion: easm.unguess.io/v1
+itemFamily: domains
+type: domain
+syncable: true
+mappings:
+  identifier: "{{ .id | sha256sum }}"
+  metadata:
+    title: "{{ get \"domain\" . .id | quote }}"
+  spec:
+    domain: "{{ get \"domain\" . \"\" | quote }}"
+    dns: "{{ get \"dns\" . (object) | toJSON }}"
+    whois: "{{ get \"whois\" . (object) | toJSON }}"
+    zonetransfer: "{{ get \"zonetransfer\" . (list) | toJSON }}"
+    azure: "{{ get \"azure\" . (object) | toJSON }}"
+    misconfig: "{{ get \"misconfig\" . (list) | toJSON }}"
diff --git a/docs/external-sources/easm/mappings/endpoints.yaml b/docs/external-sources/easm/mappings/endpoints.yaml
new file mode 100644
index 0000000..a413793
--- /dev/null
+++ b/docs/external-sources/easm/mappings/endpoints.yaml
@@ -0,0 +1,13 @@
+apiVersion: easm.unguess.io/v1
+itemFamily: endpoints
+type: endpoint
+syncable: true
+mappings:
+  identifier: "{{ .id | sha256sum }}"
+  metadata:
+    title: "{{ .id | quote }}"
+  spec:
+    host: "{{ get \"host\" . \"\" | quote }}"
+    port: "{{ get \"port\" . 0 }}"
+    http: "{{ get \"http\" . (object) | toJSON }}"
+    tls: "{{ get \"tls\" . (object) | toJSON }}"
diff --git a/docs/external-sources/easm/mappings/hosts.yaml b/docs/external-sources/easm/mappings/hosts.yaml
new file mode 100644
index 0000000..c6bbed5
--- /dev/null
+++ b/docs/external-sources/easm/mappings/hosts.yaml
@@ -0,0 +1,14 @@
+apiVersion: easm.unguess.io/v1
+itemFamily: hosts
+type: host
+syncable: true
+mappings:
+  identifier: "{{ .id | sha256sum }}"
+  metadata:
+    title: "{{ get \"fqdn\" . .id | quote }}"
+  spec:
+    fqdn: "{{ get \"fqdn\" . \"\" | quote }}"
+    status_code: "{{ get \"status_code\" . \"\" | quote }}"
+    a: "{{ get \"a\" . (list) | toJSON }}"
+    aaaa: "{{ get \"aaaa\" . (list) | toJSON }}"
+    cname: "{{ get \"cname\" . (list) | toJSON }}"
diff --git a/docs/external-sources/easm/mappings/ips.yaml b/docs/external-sources/easm/mappings/ips.yaml
new file mode 100644
index 0000000..e04e6e2
--- /dev/null
+++ b/docs/external-sources/easm/mappings/ips.yaml
@@ -0,0 +1,13 @@
+apiVersion: easm.unguess.io/v1
+itemFamily: ips
+type: ip
+syncable: true
+mappings:
+  identifier: "{{ .id | sha256sum }}"
+  metadata:
+    title: "{{ get \"ip\" . .id | quote }}"
+  spec:
+    ip: "{{ get \"ip\" . \"\" | quote }}"
+    ptr: "{{ get \"ptr\" . \"\" | quote }}"
+    ports: "{{ get \"ports\" . (list) | toJSON }}"
+    geo: "{{ get \"geo\" . (object) | toJSON }}"
diff --git a/docs/external-sources/easm/mappings/vulnerabilities.yaml b/docs/external-sources/easm/mappings/vulnerabilities.yaml
new file mode 100644
index 0000000..fd21fe7
--- /dev/null
+++ b/docs/external-sources/easm/mappings/vulnerabilities.yaml
@@ -0,0 +1,20 @@
+apiVersion: easm.unguess.io/v1
+itemFamily: vulnerabilities
+type: vulnerability
+syncable: true
+mappings:
+  identifier: "{{ .id | sha256sum }}"
+  metadata:
+    title: "{{ get \"name\" . .id | quote }}"
+  spec:
+    template_id: "{{ get \"template_id\" . \"\" | quote }}"
+    name: "{{ get \"name\" . \"\" | quote }}"
+    severity: "{{ get \"severity\" . \"\" | quote }}"
+    target: "{{ get \"target\" . \"\" | quote }}"
+    description: "{{ get \"description\" . \"\" | quote }}"
+    tags: "{{ get \"tags\" . (list) | toJSON }}"
+    extracted_results: "{{ get \"extracted_results\" . (list) | toJSON }}"
+    matcher_name: "{{ get \"matcher_name\" . \"\" | quote }}"
+    scheme: "{{ get \"scheme\" . \"\" | quote }}"
+    meta: "{{ get \"meta\" . (object) | toJSON }}"
+    owasp: "{{ get \"owasp\" . (object) | toJSON }}"
diff --git a/internal/cmd/common.go b/internal/cmd/common.go
index b0a1fb6..51f02c0 100644
--- a/internal/cmd/common.go
+++ b/internal/cmd/common.go
@@ -20,6 +20,7 @@ import (
 	azuredevops "github.com/mia-platform/ibdm/internal/source/azure-devops"
 	"github.com/mia-platform/ibdm/internal/source/bitbucket"
 	"github.com/mia-platform/ibdm/internal/source/console"
+	"github.com/mia-platform/ibdm/internal/source/easm"
 	"github.com/mia-platform/ibdm/internal/source/gcp"
 	"github.com/mia-platform/ibdm/internal/source/github"
 	"github.com/mia-platform/ibdm/internal/source/gitlab"
@@ -36,6 +37,8 @@ const (
 	bitbucketDescription   = "Bitbucket integration"
 	consoleSource          = "console"
 	consoleDescription     = "Mia Platform Console integration"
+	easmSource             = "easm"
+	easmDescription        = "External Attack Surface Management (EASM) integration"
 	gcpSource              = "gcp"
 	gcpDescription         = "Google Cloud Platform integration"
 	githubSource           = "github"
@@ -70,6 +73,7 @@ var (
 		azureSource:       azureDescription,
 		bitbucketSource:   bitbucketDescription,
 		consoleSource:     consoleDescription,
+		easmSource:        easmDescription,
 		gcpSource:         gcpDescription,
 		githubSource:      githubDescription,
 		gitlabSource:      gitlabDescription,
@@ -127,6 +131,8 @@ func sourceFromIntegrationName(integrationName string) (any, error) {
 		return github.NewSource()
 	case consoleSource:
 		return console.NewSource()
+	case easmSource:
+		return easm.NewSource()
 	case gitlabSource:
 		return gitlab.NewSource()
 	case nexusSource:
diff --git a/internal/source/easm/client.go b/internal/source/easm/client.go
new file mode 100644
index 0000000..ab105d8
--- /dev/null
+++ b/internal/source/easm/client.go
@@ -0,0 +1,114 @@
+// Copyright Mia srl
+// SPDX-License-Identifier: AGPL-3.0-only or Commercial
+
+package easm
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+)
+
+const (
+	// maxErrorBodySize limits how many bytes we read from error response bodies
+	// to avoid unbounded memory allocation on unexpectedly large payloads.
+	maxErrorBodySize = 1024
+
+	// nextCursorHeader carries the cursor for the next page; empty or absent on the last page.
+	nextCursorHeader = "X-Next-Cursor"
+	// cursorQueryParam names the query parameter used to request a specific page.
+	cursorQueryParam = "cursor"
+)
+
+// client wraps an HTTP client with EASM endpoint configuration.
+type client struct {
+	baseURL  *url.URL
+	dataPath string
+	customer string
+	token    string
+
+	httpClient *http.Client
+}
+
+// newClient creates a client from the given config.
+func newClient(cfg config) (*client, error) {
+	u, err := url.Parse(cfg.BaseURL)
+	if err != nil {
+		return nil, fmt.Errorf("%w: invalid EASM_BASE_URL: %w", ErrInvalidEnvVariable, err)
+	}
+
+	return &client{
+		baseURL:  u,
+		dataPath: cfg.DataPath,
+		customer: cfg.Customer,
+		token:    cfg.Token,
+		httpClient: &http.Client{
+			Timeout: cfg.HTTPTimeout,
+		},
+	}, nil
+}
+
+// dataPage is a single page of the /data response: a flat list of items plus
+// the cursor for the next page (empty when this is the last page).
+type dataPage struct {
+	items      []map[string]any
+	nextCursor string
+}
+
+// fetchDataPage retrieves a single page of items from the endpoint. Pass an
+// empty cursor for the first page. "Latest completed run" is resolved
+// server-side; the client never sees or picks a run id.
+func (c *client) fetchDataPage(ctx context.Context, cursor string) (*dataPage, error) {
+	u := c.baseURL.JoinPath(c.dataPath)
+	if cursor != "" {
+		q := u.Query()
+		q.Set(cursorQueryParam, cursor)
+		u.RawQuery = q.Encode()
+	}
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil)
+	if err != nil {
+		return nil, err
+	}
+
+	req.Header.Set("Accept", "application/json")
+	// X-Customer scopes the request to a single customer and is always set.
+	// The bearer token authenticates the caller and is sent only once
+	// configured — the backend has no auth yet.
+	if c.token != "" {
+		req.Header.Set("Authorization", "Bearer "+c.token)
+	}
+	if c.customer != "" {
+		req.Header.Set("X-Customer", c.customer)
+	}
+
+	resp, err := c.httpClient.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return nil, readErrorResponse(resp)
+	}
+
+	var items []map[string]any
+	if err := json.NewDecoder(resp.Body).Decode(&items); err != nil {
+		return nil, fmt.Errorf("failed to decode data response: %w", err)
+	}
+
+	return &dataPage{
+		items:      items,
+		nextCursor: resp.Header.Get(nextCursorHeader),
+	}, nil
+}
+
+// readErrorResponse constructs an error from a non-2xx HTTP response,
+// reading at most maxErrorBodySize bytes from the body.
+func readErrorResponse(resp *http.Response) error {
+	body, _ := io.ReadAll(io.LimitReader(resp.Body, maxErrorBodySize))
+	return fmt.Errorf("easm API returned status %d: %s", resp.StatusCode, string(body))
+}
diff --git a/internal/source/easm/client_test.go b/internal/source/easm/client_test.go
new file mode 100644
index 0000000..8b25a8b
--- /dev/null
+++ b/internal/source/easm/client_test.go
@@ -0,0 +1,208 @@
+// Copyright Mia srl
+// SPDX-License-Identifier: AGPL-3.0-only or Commercial
+
+package easm
+
+import (
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewClient(t *testing.T) {
+	t.Parallel()
+
+	testCases := map[string]struct {
+		config       config
+		expectErr    error
+		assertClient func(t *testing.T, c *client)
+	}{
+		"valid base URL": {
+			config: config{
+				BaseURL:     "https://easm.example.com",
+				DataPath:    "/data",
+				Customer:    "acme",
+				Token:       "test-token",
+				HTTPTimeout: 5 * time.Second,
+			},
+			assertClient: func(t *testing.T, c *client) {
+				t.Helper()
+				assert.Equal(t, "https://easm.example.com", c.baseURL.String())
+				assert.Equal(t, "/data", c.dataPath)
+				assert.Equal(t, "acme", c.customer)
+				assert.Equal(t, "test-token", c.token)
+				assert.Equal(t, 5*time.Second, c.httpClient.Timeout)
+			},
+		},
+		"invalid base URL": {
+			config: config{
+				BaseURL: "://invalid",
+			},
+			expectErr: ErrInvalidEnvVariable,
+		},
+	}
+
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+
+			c, err := newClient(tc.config)
+			if tc.expectErr != nil {
+				require.ErrorIs(t, err, tc.expectErr)
+				return
+			}
+
+			require.NoError(t, err)
+			tc.assertClient(t, c)
+		})
+	}
+}
+
+func TestFetchDataPagePagination(t *testing.T) {
+	t.Parallel()
+
+	testCases := map[string]struct {
+		body           string
+		nextCursor     string
+		expectedItems  int
+		expectedCursor string
+	}{
+		"first page with next cursor": {
+			body:           `[{"id":"1","type":"domain"},{"id":"2","type":"host"}]`,
+			nextCursor:     "cursor-2",
+			expectedItems:  2,
+			expectedCursor: "cursor-2",
+		},
+		"last page without cursor": {
+			body:           `[{"id":"3","type":"ip"}]`,
+			nextCursor:     "",
+			expectedItems:  1,
+			expectedCursor: "",
+		},
+		"empty page": {
+			body:           `[]`,
+			nextCursor:     "",
+			expectedItems:  0,
+			expectedCursor: "",
+		},
+	}
+
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+
+			handler := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+				if tc.nextCursor != "" {
+					w.Header().Set(nextCursorHeader, tc.nextCursor)
+				}
+				w.Header().Set("Content-Type", "application/json")
+				_, _ = w.Write([]byte(tc.body))
+			})
+
+			source := newTestSource(t, handler)
+
+			page, err := source.client.fetchDataPage(t.Context(), "")
+			require.NoError(t, err)
+			assert.Len(t, page.items, tc.expectedItems)
+			assert.Equal(t, tc.expectedCursor, page.nextCursor)
+		})
+	}
+}
+
+func TestFetchDataPageSendsCursor(t *testing.T) {
+	t.Parallel()
+
+	var gotCursor string
+	handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		gotCursor = r.URL.Query().Get(cursorQueryParam)
+		_, _ = w.Write([]byte(`[]`))
+	})
+
+	source := newTestSource(t, handler)
+
+	_, err := source.client.fetchDataPage(t.Context(), "cursor-42")
+	require.NoError(t, err)
+	assert.Equal(t, "cursor-42", gotCursor)
+}
+
+func TestFetchDataPageHeaders(t *testing.T) {
+	t.Parallel()
+
+	t.Run("token and customer set", func(t *testing.T) {
+		t.Parallel()
+
+		var gotReq *http.Request
+		handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			gotReq = r.Clone(r.Context())
+			_, _ = w.Write([]byte(`[]`))
+		})
+		source := newTestSource(t, handler)
+
+		_, err := source.client.fetchDataPage(t.Context(), "")
+		require.NoError(t, err)
+		assert.Equal(t, "application/json", gotReq.Header.Get("Accept"))
+		assert.Equal(t, "Bearer test-token", gotReq.Header.Get("Authorization"))
+		assert.Equal(t, "acme", gotReq.Header.Get("X-Customer"))
+	})
+
+	t.Run("empty token omits Authorization header", func(t *testing.T) {
+		t.Parallel()
+
+		var gotReq *http.Request
+		handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			gotReq = r.Clone(r.Context())
+			_, _ = w.Write([]byte(`[]`))
+		})
+		source := newTestSource(t, handler)
+		// Build a client without a token pointing at the same server.
+		c := &client{
+			baseURL:    source.client.baseURL,
+			dataPath:   source.client.dataPath,
+			customer:   source.client.customer,
+			token:      "",
+			httpClient: source.client.httpClient,
+		}
+
+		_, err := c.fetchDataPage(t.Context(), "")
+		require.NoError(t, err)
+		assert.Empty(t, gotReq.Header.Get("Authorization"))
+		assert.Equal(t, "acme", gotReq.Header.Get("X-Customer"))
+	})
+}
+
+func TestFetchDataPageErrors(t *testing.T) {
+	t.Parallel()
+
+	t.Run("non-200 status", func(t *testing.T) {
+		t.Parallel()
+
+		handler := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+			w.WriteHeader(http.StatusInternalServerError)
+			_, _ = w.Write([]byte("boom"))
+		})
+
+		source := newTestSource(t, handler)
+
+		_, err := source.client.fetchDataPage(t.Context(), "")
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "500")
+		assert.Contains(t, err.Error(), "boom")
+	})
+
+	t.Run("malformed JSON body", func(t *testing.T) {
+		t.Parallel()
+
+		handler := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+			_, _ = w.Write([]byte("not json"))
+		})
+
+		source := newTestSource(t, handler)
+
+		_, err := source.client.fetchDataPage(t.Context(), "")
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "failed to decode data response")
+	})
+}
diff --git a/internal/source/easm/config.go b/internal/source/easm/config.go
new file mode 100644
index 0000000..fe3955c
--- /dev/null
+++ b/internal/source/easm/config.go
@@ -0,0 +1,72 @@
+// Copyright Mia srl
+// SPDX-License-Identifier: AGPL-3.0-only or Commercial
+
+package easm
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/caarlos0/env/v11"
+)
+
+var (
+	// ErrMissingEnvVariable reports missing mandatory environment variables.
+	ErrMissingEnvVariable = errors.New("missing environment variable")
+	// ErrInvalidEnvVariable reports malformed environment variable values.
+	ErrInvalidEnvVariable = errors.New("invalid environment value")
+)
+
+// config holds the environment-driven EASM settings.
+type config struct {
+	// BaseURL is the FE backend base URL, e.g. http://localhost:8000 (mock) or the product backend (prod).
+	BaseURL string `env:"EASM_BASE_URL"`
+	// DataPath is the path of the read endpoint appended to BaseURL.
+	DataPath string `env:"EASM_DATA_PATH" envDefault:"/data"`
+	// Customer scopes the request to a single customer via the X-Customer
+	// header. Always required: it selects whose scan results to read.
+	Customer string `env:"EASM_CUSTOMER"`
+	// Token authenticates the caller to the backend via Authorization: Bearer.
+	// Optional: when set, it is sent as the Authorization: Bearer header.
+	Token string `env:"EASM_TOKEN"`
+	// HTTPTimeout bounds each request to the endpoint.
+	HTTPTimeout time.Duration `env:"EASM_HTTP_TIMEOUT" envDefault:"30s"`
+}
+
+// loadConfigFromEnv parses environment variables into a config struct and
+// validates that all required fields are present.
+func loadConfigFromEnv() (config, error) {
+	cfg, err := env.ParseAs[config]()
+	if err != nil {
+		return config{}, err
+	}
+
+	if err := validateConfig(cfg); err != nil {
+		return config{}, err
+	}
+
+	return cfg, nil
+}
+
+// validateConfig checks that the required config fields are non-empty. Customer
+// is always required: it scopes the request to a single customer via the
+// X-Customer header. Token is optional and, when set, authenticates the caller
+// via Authorization: Bearer.
+func validateConfig(cfg config) error {
+	missing := make([]string, 0)
+
+	if cfg.BaseURL == "" {
+		missing = append(missing, "EASM_BASE_URL")
+	}
+	if cfg.Customer == "" {
+		missing = append(missing, "EASM_CUSTOMER")
+	}
+
+	if len(missing) > 0 {
+		return fmt.Errorf("%w: %s", ErrMissingEnvVariable, strings.Join(missing, ", "))
+	}
+
+	return nil
+}
diff --git a/internal/source/easm/config_test.go b/internal/source/easm/config_test.go
new file mode 100644
index 0000000..6ce2664
--- /dev/null
+++ b/internal/source/easm/config_test.go
@@ -0,0 +1,132 @@
+// Copyright Mia srl
+// SPDX-License-Identifier: AGPL-3.0-only or Commercial
+
+package easm
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestLoadConfigFromEnv(t *testing.T) {
+	testCases := map[string]struct {
+		setupEnv       func(t *testing.T)
+		expectedConfig config
+		expectedErr    error
+	}{
+		"required env vars set, defaults applied": {
+			setupEnv: func(t *testing.T) {
+				t.Helper()
+				t.Setenv("EASM_BASE_URL", "https://easm.example.com")
+				t.Setenv("EASM_CUSTOMER", "acme")
+			},
+			expectedConfig: config{
+				BaseURL:     "https://easm.example.com",
+				DataPath:    "/data",
+				Customer:    "acme",
+				HTTPTimeout: 30 * time.Second,
+			},
+		},
+		"all env vars set including optional": {
+			setupEnv: func(t *testing.T) {
+				t.Helper()
+				t.Setenv("EASM_BASE_URL", "https://easm.example.com")
+				t.Setenv("EASM_CUSTOMER", "acme")
+				t.Setenv("EASM_TOKEN", "secret")
+				t.Setenv("EASM_HTTP_TIMEOUT", "10s")
+			},
+			expectedConfig: config{
+				BaseURL:     "https://easm.example.com",
+				DataPath:    "/data",
+				Customer:    "acme",
+				Token:       "secret",
+				HTTPTimeout: 10 * time.Second,
+			},
+		},
+		"missing EASM_BASE_URL": {
+			setupEnv: func(t *testing.T) {
+				t.Helper()
+				t.Setenv("EASM_CUSTOMER", "acme")
+			},
+			expectedErr: ErrMissingEnvVariable,
+		},
+		"missing EASM_CUSTOMER": {
+			setupEnv: func(t *testing.T) {
+				t.Helper()
+				t.Setenv("EASM_BASE_URL", "https://easm.example.com")
+			},
+			expectedErr: ErrMissingEnvVariable,
+		},
+		"missing all required vars": {
+			setupEnv: func(t *testing.T) {
+				t.Helper()
+			},
+			expectedErr: ErrMissingEnvVariable,
+		},
+	}
+
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			tc.setupEnv(t)
+
+			cfg, err := loadConfigFromEnv()
+			if tc.expectedErr != nil {
+				require.ErrorIs(t, err, tc.expectedErr)
+				return
+			}
+
+			require.NoError(t, err)
+			assert.Equal(t, tc.expectedConfig, cfg)
+		})
+	}
+}
+
+func TestValidateConfig(t *testing.T) {
+	t.Parallel()
+
+	testCases := map[string]struct {
+		config    config
+		expectErr error
+	}{
+		"valid config": {
+			config: config{
+				BaseURL:     "https://easm.example.com",
+				DataPath:    "/data",
+				Customer:    "acme",
+				HTTPTimeout: 30 * time.Second,
+			},
+		},
+		"missing base URL": {
+			config: config{
+				Customer: "acme",
+			},
+			expectErr: ErrMissingEnvVariable,
+		},
+		"missing customer": {
+			config: config{
+				BaseURL: "https://easm.example.com",
+			},
+			expectErr: ErrMissingEnvVariable,
+		},
+		"all missing": {
+			config:    config{},
+			expectErr: ErrMissingEnvVariable,
+		},
+	}
+
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+
+			err := validateConfig(tc.config)
+			if tc.expectErr != nil {
+				assert.ErrorIs(t, err, tc.expectErr)
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
+}
diff --git a/internal/source/easm/doc.go b/internal/source/easm/doc.go
new file mode 100644
index 0000000..3606d4f
--- /dev/null
+++ b/internal/source/easm/doc.go
@@ -0,0 +1,8 @@
+// Copyright Mia srl
+// SPDX-License-Identifier: AGPL-3.0-only or Commercial
+
+// Package easm provides a source implementation that integrates EASM scan
+// results into the Catalog. It reads the customer's latest completed run from
+// the product backend's /data endpoint as a single paginated list of items,
+// each tagged with a "type" discriminator, and emits one source.Data per item.
+package easm
diff --git a/internal/source/easm/helper_test.go b/internal/source/easm/helper_test.go
new file mode 100644
index 0000000..8cc8b88
--- /dev/null
+++ b/internal/source/easm/helper_test.go
@@ -0,0 +1,69 @@
+// Copyright Mia srl
+// SPDX-License-Identifier: AGPL-3.0-only or Commercial
+
+package easm
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/mia-platform/ibdm/internal/source"
+)
+
+var testTime = time.Date(2025, time.March, 1, 12, 0, 0, 0, time.UTC)
+
+func init() {
+	timeSource = func() time.Time {
+		return testTime
+	}
+}
+
+// newTestSource builds a Source whose client points at an httptest server
+// running the given handler. The server is closed automatically when the test
+// finishes.
+func newTestSource(t *testing.T, handler http.Handler) *Source {
+	t.Helper()
+
+	server := httptest.NewServer(handler)
+	t.Cleanup(server.Close)
+
+	u, err := url.Parse(server.URL)
+	require.NoError(t, err)
+
+	cfg := config{
+		BaseURL:     server.URL,
+		DataPath:    "/data",
+		Customer:    "acme",
+		Token:       "test-token",
+		HTTPTimeout: 5 * time.Second,
+	}
+
+	return &Source{
+		config: cfg,
+		client: &client{
+			baseURL:  u,
+			dataPath: cfg.DataPath,
+			customer: cfg.Customer,
+			token:    cfg.Token,
+			httpClient: &http.Client{
+				Timeout: cfg.HTTPTimeout,
+			},
+		},
+	}
+}
+
+// collectData drains a source.Data channel into a slice.
+func collectData(t *testing.T, ch <-chan source.Data) []source.Data {
+	t.Helper()
+
+	var result []source.Data
+	for d := range ch {
+		result = append(result, d)
+	}
+	return result
+}
diff --git a/internal/source/easm/source.go b/internal/source/easm/source.go
new file mode 100644
index 0000000..657b651
--- /dev/null
+++ b/internal/source/easm/source.go
@@ -0,0 +1,69 @@
+// Copyright Mia srl
+// SPDX-License-Identifier: AGPL-3.0-only or Commercial
+
+package easm
+
+import (
+	"errors"
+	"sync"
+	"time"
+
+	"github.com/mia-platform/ibdm/internal/source"
+)
+
+const (
+	loggerName = "ibdm:source:easm"
+
+	domainType        = "domain"
+	hostType          = "host"
+	ipType            = "ip"
+	endpointType      = "endpoint"
+	vulnerabilityType = "vulnerability"
+)
+
+// knownTypes is the set of item types the source can emit. The endpoint tags
+// each item with one of these; anything else is skipped.
+var knownTypes = map[string]struct{}{
+	domainType:        {},
+	hostType:          {},
+	ipType:            {},
+	endpointType:      {},
+	vulnerabilityType: {},
+}
+
+var (
+	// ErrEASMSource wraps errors emitted by the EASM source implementation.
+	ErrEASMSource = errors.New("easm source")
+
+	// timeSource is a replaceable function for obtaining the current time.
+	// Tests override this to produce deterministic timestamps.
+	timeSource = time.Now
+)
+
+var _ source.SyncableSource = &Source{}
+
+// Source implements source.SyncableSource for our EASM scan results.
+type Source struct {
+	config config
+	client *client
+
+	syncLock sync.Mutex
+}
+
+// NewSource creates a new EASM Source reading configuration from environment variables.
+func NewSource() (*Source, error) {
+	cfg, err := loadConfigFromEnv()
+	if err != nil {
+		return nil, handleErr(err)
+	}
+
+	c, err := newClient(cfg)
+	if err != nil {
+		return nil, handleErr(err)
+	}
+
+	return &Source{
+		config: cfg,
+		client: c,
+	}, nil
+}
diff --git a/internal/source/easm/source_test.go b/internal/source/easm/source_test.go
new file mode 100644
index 0000000..f930ecb
--- /dev/null
+++ b/internal/source/easm/source_test.go
@@ -0,0 +1,68 @@
+// Copyright Mia srl
+// SPDX-License-Identifier: AGPL-3.0-only or Commercial
+
+package easm
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewSource(t *testing.T) {
+	testCases := map[string]struct {
+		setupEnv    func(t *testing.T)
+		expectErrs  []error
+		assertValid func(t *testing.T, s *Source)
+	}{
+		"valid env": {
+			setupEnv: func(t *testing.T) {
+				t.Helper()
+				t.Setenv("EASM_BASE_URL", "https://easm.example.com")
+				t.Setenv("EASM_CUSTOMER", "acme")
+			},
+			assertValid: func(t *testing.T, s *Source) {
+				t.Helper()
+				assert.Equal(t, "https://easm.example.com", s.config.BaseURL)
+				assert.Equal(t, "acme", s.config.Customer)
+				assert.NotNil(t, s.client)
+			},
+		},
+		"config error: missing required var": {
+			setupEnv: func(t *testing.T) {
+				t.Helper()
+				t.Setenv("EASM_CUSTOMER", "acme")
+			},
+			expectErrs: []error{ErrEASMSource, ErrMissingEnvVariable},
+		},
+		"client error: invalid base URL": {
+			setupEnv: func(t *testing.T) {
+				t.Helper()
+				t.Setenv("EASM_BASE_URL", "://invalid")
+				t.Setenv("EASM_CUSTOMER", "acme")
+			},
+			expectErrs: []error{ErrEASMSource, ErrInvalidEnvVariable},
+		},
+	}
+
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			tc.setupEnv(t)
+
+			s, err := NewSource()
+			if len(tc.expectErrs) > 0 {
+				require.Error(t, err)
+				for _, target := range tc.expectErrs {
+					assert.ErrorIs(t, err, target)
+				}
+				assert.Nil(t, s)
+				return
+			}
+
+			require.NoError(t, err)
+			require.NotNil(t, s)
+			tc.assertValid(t, s)
+		})
+	}
+}
diff --git a/internal/source/easm/sync.go b/internal/source/easm/sync.go
new file mode 100644
index 0000000..cf5dcd6
--- /dev/null
+++ b/internal/source/easm/sync.go
@@ -0,0 +1,88 @@
+// Copyright Mia srl
+// SPDX-License-Identifier: AGPL-3.0-only or Commercial
+
+package easm
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/mia-platform/ibdm/internal/logger"
+	"github.com/mia-platform/ibdm/internal/source"
+)
+
+// StartSyncProcess implements source.SyncableSource. It reads the customer's
+// latest completed run from the /data endpoint as a single cursor-paginated
+// list and emits one source.Data per item, routed by the item's own "type"
+// field. Filtering by typesToSync lets the pipeline sync a subset of types.
+func (s *Source) StartSyncProcess(ctx context.Context, typesToSync map[string]source.Extra, results chan<- source.Data) error {
+	log := logger.FromContext(ctx).WithName(loggerName)
+
+	if !s.syncLock.TryLock() {
+		log.Debug("sync process already running")
+		return nil
+	}
+	defer s.syncLock.Unlock()
+
+	// Log unknown requested types.
+	for typeKey := range typesToSync {
+		if _, ok := knownTypes[typeKey]; !ok {
+			log.Debug("unknown type requested, skipping", "type", typeKey)
+		}
+	}
+
+	cursor := ""
+	for {
+		if err := ctx.Err(); err != nil {
+			return nil
+		}
+
+		log.Trace("fetching data page", "cursor", cursor)
+
+		page, err := s.client.fetchDataPage(ctx, cursor)
+		if err != nil {
+			return handleErr(err)
+		}
+
+		for _, item := range page.items {
+			itemType, ok := item["type"].(string)
+			if !ok || itemType == "" {
+				log.Debug("item without a type, skipping", "id", item["id"])
+				continue
+			}
+
+			if _, requested := typesToSync[itemType]; !requested {
+				continue
+			}
+
+			results <- source.Data{
+				Type:      itemType,
+				Operation: source.DataOperationUpsert,
+				Values:    item,
+				Time:      timeSource(),
+			}
+		}
+
+		if page.nextCursor == "" {
+			break
+		}
+		cursor = page.nextCursor
+	}
+
+	return nil
+}
+
+// handleErr wraps non-nil errors with ErrEASMSource, matching the project convention.
+// Context cancellation errors are silently swallowed (return nil).
+func handleErr(err error) error {
+	if err == nil {
+		return nil
+	}
+
+	if errors.Is(err, context.Canceled) {
+		return nil
+	}
+
+	return fmt.Errorf("%w: %w", ErrEASMSource, err)
+}
diff --git a/internal/source/easm/sync_test.go b/internal/source/easm/sync_test.go
new file mode 100644
index 0000000..048127e
--- /dev/null
+++ b/internal/source/easm/sync_test.go
@@ -0,0 +1,247 @@
+// Copyright Mia srl
+// SPDX-License-Identifier: AGPL-3.0-only or Commercial
+
+package easm
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/mia-platform/ibdm/internal/source"
+)
+
+// runSync drives StartSyncProcess to completion and returns the emitted data
+// and the returned error.
+func runSync(t *testing.T, s *Source, typesToSync map[string]source.Extra) ([]source.Data, error) {
+	t.Helper()
+
+	ch := make(chan source.Data, 100)
+	var data []source.Data
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		data = collectData(t, ch)
+	}()
+
+	err := s.StartSyncProcess(t.Context(), typesToSync, ch)
+	close(ch)
+	<-done
+
+	return data, err
+}
+
+// writeItems encodes a JSON array of items as the /data response body.
+func writeItems(w http.ResponseWriter, items []map[string]any) {
+	w.Header().Set("Content-Type", "application/json")
+	_ = json.NewEncoder(w).Encode(items)
+}
+
+func TestStartSyncProcessRouting(t *testing.T) {
+	t.Parallel()
+
+	items := []map[string]any{
+		{"id": "1", "type": domainType},
+		{"id": "2", "type": hostType},
+		{"id": "3", "type": ipType},
+		{"id": "4", "type": endpointType},
+	}
+
+	handler := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		writeItems(w, items)
+	})
+	s := newTestSource(t, handler)
+
+	typesToSync := map[string]source.Extra{
+		domainType: {},
+		ipType:     {},
+	}
+
+	data, err := runSync(t, s, typesToSync)
+	require.NoError(t, err)
+
+	// Only domain and ip are requested; host and endpoint are filtered out.
+	require.Len(t, data, 2)
+	for _, d := range data {
+		assert.Equal(t, source.DataOperationUpsert, d.Operation)
+		assert.Equal(t, testTime, d.Time)
+	}
+
+	expected := []source.Data{
+		{Type: domainType, Operation: source.DataOperationUpsert, Values: items[0], Time: testTime},
+		{Type: ipType, Operation: source.DataOperationUpsert, Values: items[2], Time: testTime},
+	}
+	assert.ElementsMatch(t, expected, data)
+}
+
+func TestStartSyncProcessPagination(t *testing.T) {
+	t.Parallel()
+
+	page1 := []map[string]any{
+		{"id": "1", "type": domainType},
+		{"id": "2", "type": hostType},
+	}
+	page2 := []map[string]any{
+		{"id": "3", "type": ipType},
+	}
+
+	handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch r.URL.Query().Get(cursorQueryParam) {
+		case "":
+			w.Header().Set(nextCursorHeader, "page2")
+			writeItems(w, page1)
+		case "page2":
+			writeItems(w, page2)
+		default:
+			t.Errorf("unexpected cursor %q", r.URL.Query().Get(cursorQueryParam))
+		}
+	})
+	s := newTestSource(t, handler)
+
+	typesToSync := map[string]source.Extra{
+		domainType: {},
+		hostType:   {},
+		ipType:     {},
+	}
+
+	data, err := runSync(t, s, typesToSync)
+	require.NoError(t, err)
+
+	// All three items, across both pages, are collected.
+	require.Len(t, data, 3)
+	expected := []source.Data{
+		{Type: domainType, Operation: source.DataOperationUpsert, Values: page1[0], Time: testTime},
+		{Type: hostType, Operation: source.DataOperationUpsert, Values: page1[1], Time: testTime},
+		{Type: ipType, Operation: source.DataOperationUpsert, Values: page2[0], Time: testTime},
+	}
+	assert.ElementsMatch(t, expected, data)
+}
+
+func TestStartSyncProcessSkipsItemsWithoutType(t *testing.T) {
+	t.Parallel()
+
+	items := []map[string]any{
+		{"id": "1"},                     // no type field
+		{"id": "2", "type": ""},         // empty type
+		{"id": "3", "type": domainType}, // valid
+	}
+
+	handler := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		writeItems(w, items)
+	})
+	s := newTestSource(t, handler)
+
+	data, err := runSync(t, s, map[string]source.Extra{domainType: {}})
+	require.NoError(t, err)
+
+	require.Len(t, data, 1)
+	assert.Equal(t, domainType, data[0].Type)
+	assert.Equal(t, items[2], data[0].Values)
+}
+
+func TestStartSyncProcessFetchError(t *testing.T) {
+	t.Parallel()
+
+	handler := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusInternalServerError)
+		_, _ = w.Write([]byte("boom"))
+	})
+	s := newTestSource(t, handler)
+
+	data, err := runSync(t, s, map[string]source.Extra{domainType: {}})
+	require.Error(t, err)
+	assert.ErrorIs(t, err, ErrEASMSource)
+	assert.Empty(t, data)
+}
+
+func TestStartSyncProcessAlreadyRunning(t *testing.T) {
+	t.Parallel()
+
+	handler := http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) {
+		t.Fatal("no request should be made when the sync lock is already held")
+	})
+	s := newTestSource(t, handler)
+
+	// Simulate an in-flight sync by holding the lock.
+	s.syncLock.Lock()
+	defer s.syncLock.Unlock()
+
+	data, err := runSync(t, s, map[string]source.Extra{domainType: {}})
+	assert.NoError(t, err)
+	assert.Empty(t, data)
+}
+
+func TestContextCancellationInSync(t *testing.T) {
+	t.Parallel()
+
+	// Handler always advertises another page, so the loop would run forever
+	// unless the context cancellation breaks it.
+	handler := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set(nextCursorHeader, "next")
+		writeItems(w, []map[string]any{{"id": "1", "type": domainType}})
+	})
+	s := newTestSource(t, handler)
+
+	ctx, cancel := context.WithCancel(t.Context())
+
+	ch := make(chan source.Data, 100)
+	done := make(chan error, 1)
+	go func() {
+		done <- s.StartSyncProcess(ctx, map[string]source.Extra{domainType: {}}, ch)
+		close(ch)
+	}()
+
+	// Read one item, then cancel.
+	<-ch
+	cancel()
+
+	err := <-done
+	assert.NoError(t, err)
+}
+
+func TestHandleErr(t *testing.T) {
+	t.Parallel()
+
+	testCases := map[string]struct {
+		err       error
+		expectNil bool
+	}{
+		"nil error": {
+			err:       nil,
+			expectNil: true,
+		},
+		"context canceled": {
+			err:       context.Canceled,
+			expectNil: true,
+		},
+		"wrapped context canceled": {
+			err:       fmt.Errorf("fetch failed: %w", context.Canceled),
+			expectNil: true,
+		},
+		"regular error": {
+			err: errors.New("something failed"),
+		},
+	}
+
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+
+			result := handleErr(tc.err)
+			if tc.expectNil {
+				assert.NoError(t, result)
+				return
+			}
+
+			require.Error(t, result)
+			assert.ErrorIs(t, result, ErrEASMSource)
+		})
+	}
+}