How to monitor public app stores for brand impersonation signals
Build a defensible public-surface watchlist for lookalike apps and brand abuse: collect candidates, preserve evidence, score signals, monitor change, and send humans only the cases worth reviewing.
Brand impersonation is rarely discovered by a dramatic alert. More often, support receives a screenshot of an app using a familiar name, a customer asks whether an account is real, or an ad campaign appears before the security team has seen it. The useful response is not “we scraped the internet.” It is a reviewable case that answers four questions:
- Is this an official asset or a candidate for review?
- What did it say, where did we see it, and when?
- Is it new, or did it change since the last check?
- Which team owns the next action?
ScrapeNest covers the public-web collection part: collect public app-store results, preserve the page and screenshot as evidence, and watch a search page for changes. It does not decide that an app is malicious, and it is not an authenticated social-media or app-store connector.
That distinction matters: a name match is a signal, not a finding.
A real public surface: Google Play search
Google Play exposes public search pages. For example:
https://play.google.com/store/search?q=openai&c=apps&hl=en
When we checked this route with ScrapeNest, it delivered public result cards. The direct ChatGPT app page also delivered the title ChatGPT. This does not label any other search result as impersonation. It proves the narrower, useful point: a public app-store search can produce candidates for a brand-protection team to inspect.
For your own watchlist, use a controlled set of terms: the brand, product names, common misspellings, and high-risk suffixes such as support, wallet, login, or verification. Keep the query set narrow enough that an analyst can review the resulting queue.
The boundary: public sources and human decisions
This guide covers public pages. Do not put usernames, passwords, session cookies, or MFA codes into scraping jobs. Login-gated pages, private groups, direct messages, and platform-restricted APIs need an authorized integration or another licensed source.
Also check the target's applicable terms and robots policy, keep request volume modest, retain only data you have a legitimate reason to use, and link reviewers back to the original listing. If a source returns a block page, 403, redirect, or a missing selector, record it as unavailable. It does not mean “no impersonation exists there.”
A complete first-run script
The script below does one complete, bounded run. It discovers candidate Google Play app URLs, opens each detail page, captures artifacts, assigns an explainable review priority, writes first-seen state to disk, and prints JSON cases to stdout.
It does not submit background jobs and then exit. That is deliberate: client.submit() only queues a job. For a first, small watchlist, client.scrape() waits for each job so the script can actually return a triage queue. Use the webhook version later when you have hundreds of candidates.
Install the SDK, set an environment variable for the key, and run it with a real brand. Never paste a live key into source control or a blog example.
pip install scrapenest-sdk
export SCRAPENEST_API_KEY="sn_live_..."
export BRAND="examplecloud"
export OFFICIAL_PACKAGE_IDS="com.examplecloud.mobile"
export OFFICIAL_DOMAINS="examplecloud.com,support.examplecloud.com"
python3 brand_watch.py
Save this as brand_watch.py:
import json
import os
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import parse_qs, urlencode, urljoin, urlparse, urlsplit
from scrapenest import ScrapeNestClient
BRAND = os.environ["BRAND"].strip()
API_KEY = os.environ["SCRAPENEST_API_KEY"]
STATE_FILE = Path(os.environ.get("BRAND_WATCH_STATE", "brand_watch_state.json"))
MAX_CANDIDATES = int(os.environ.get("MAX_CANDIDATES", "10"))
OFFICIAL_PACKAGE_IDS = {
value.strip()
for value in os.environ.get("OFFICIAL_PACKAGE_IDS", "").split(",")
if value.strip()
}
OFFICIAL_DOMAINS = {
value.strip().lower()
for value in os.environ.get("OFFICIAL_DOMAINS", "").split(",")
if value.strip()
}
def extraction_hook(client, result, hook_id):
"""Return the named extraction hook and the completed job."""
job = client.jobs.get(result.job_id)
artifact = next(
(
item
for item in job.artifacts
# Older SDK/API combinations may label this JSON artifact differently.
if item.artifact_type in {"extraction", "json"}
),
None,
)
if artifact is None:
raise RuntimeError(f"Job {result.job_id} has no extraction artifact")
payload = json.loads(client.artifacts.download_text(artifact.artifact_id))
hook = next(item for item in payload["hooks"] if item["hook_id"] == hook_id)
if hook["status"] != "succeeded":
raise RuntimeError(f"{hook_id} failed: {hook.get('error', 'unknown error')}")
if hook.get("truncated"):
raise RuntimeError(
f"{hook_id} is incomplete: {len(hook['result'])} of "
f"{hook['total_matched']} matches returned"
)
return job, hook
def package_id(app_url):
return parse_qs(urlsplit(app_url).query).get("id", [None])[0]
def load_state():
if not STATE_FILE.exists():
return {"seen": {}}
try:
return json.loads(STATE_FILE.read_text())
except json.JSONDecodeError as exc:
raise RuntimeError(f"State file is not valid JSON: {STATE_FILE}") from exc
def priority(title, app_id, outbound_links, first_seen_today):
"""Order human review. This is not a fraud or malware verdict."""
score, reasons = 0, []
name = (title or "").lower()
if BRAND.lower() in name:
score += 30
reasons.append("brand-term")
if any(word in name for word in ("support", "wallet", "login", "verify")):
score += 15
reasons.append("high-risk-term")
if first_seen_today:
score += 20
reasons.append("new-listing")
if app_id not in OFFICIAL_PACKAGE_IDS:
score += 15
reasons.append("unknown-package")
for link in outbound_links:
host = (urlparse(link).hostname or "").lower()
if host and not any(
host == domain or host.endswith("." + domain)
for domain in OFFICIAL_DOMAINS
):
score += 10
reasons.append("non-allowlisted-link")
break
return score, reasons
def main():
search_url = "https://play.google.com/store/search?" + urlencode(
{"q": BRAND, "c": "apps", "hl": "en"}
)
state = load_state()
now = datetime.now(timezone.utc).isoformat()
cases, failures = [], []
with ScrapeNestClient(api_key=API_KEY) as client:
# Standard is used here because it is the engine verified against this
# live Google Play result page. Test Light on your target before lowering it.
discovery = client.scrape(
job_type="standard",
target_url=search_url,
artifact_options={"include_extraction": True},
extraction={"hooks": [{
"hook_id": "app_links",
"type": "css",
"selector": "a[href^='/store/apps/details']",
"attribute": "href",
"all_matches": True,
"max_results": 100,
}]},
)
_, links_hook = extraction_hook(client, discovery, "app_links")
app_urls = sorted({
urljoin("https://play.google.com", href)
for href in links_hook["result"]
})[:MAX_CANDIDATES]
for app_url in app_urls:
app_id = package_id(app_url)
try:
detail = client.scrape(
job_type="standard",
target_url=app_url,
artifact_options={
"include_html": True,
"include_screenshot": True,
"include_page_metadata": True,
"include_extraction": True,
},
extraction={"hooks": [
{"hook_id": "title", "type": "css", "selector": "h1"},
{
"hook_id": "outbound_links",
"type": "css",
"selector": "a[href^='http']",
"attribute": "href",
"all_matches": True,
"max_results": 100,
},
]},
)
detail_job, title_hook = extraction_hook(client, detail, "title")
_, links_hook = extraction_hook(client, detail, "outbound_links")
title = (title_hook["result"] or [None])[0]
first_seen_today = app_id not in state["seen"]
score, reasons = priority(
title, app_id, links_hook["result"], first_seen_today
)
state["seen"].setdefault(app_id, {"first_seen": now})
cases.append({
"source": "google_play",
"source_url": app_url,
"package_id": app_id,
"title": title,
"observed_at": now,
"first_seen_today": first_seen_today,
"review_priority": score,
"signals": reasons,
"status": "needs-review",
"evidence": {
"job_id": detail.job_id,
"artifacts": {
item.artifact_type: item.artifact_id
for item in detail_job.artifacts
},
},
})
except Exception as exc:
failures.append({"source_url": app_url, "error": str(exc)})
STATE_FILE.write_text(json.dumps(state, indent=2) + "\n")
output = {
"query": search_url,
"observed_at": now,
"candidates_scanned": len(app_urls),
"cases": sorted(
cases, key=lambda case: case["review_priority"], reverse=True
),
"failures": failures,
}
print(json.dumps(output, indent=2))
if __name__ == "__main__":
main()
On its first run, every candidate is naturally new. On later runs, brand_watch_state.json retains the first-seen record, so the new-listing signal only applies to newly discovered package IDs. Replace that local file with a database before using the pattern across workers or organizations.
The script writes evidence references alongside each case: the completed job, its HTML artifact, screenshot artifact, page metadata, and extraction result. A reviewer should see what the collector saw without revisiting a page that may already have changed.
What the priority means—and what it does not
The score orders a human review queue. It must not trigger a takedown, block a user, or label an app as malicious. Every point in the example maps to an observable reason:
- a brand term in the public title;
- a high-risk term that matters to your business;
- a package ID not in the security-owned allowlist;
- a new candidate; or
- an outbound link outside approved domains.
The weights are deliberately ordinary. A financial-services company may weigh wallet more heavily; a SaaS company might add a publisher mismatch after validating a store-specific publisher extractor. Keep the rules explainable. If an analyst cannot explain a case in two sentences, the automation is too opaque.
Scale it with jobs and webhooks
The synchronous script is for a real first run, not a massive crawler. Once candidate volume justifies it, change the detail-page loop to client.submit() and collect job.completed webhooks in a durable worker. Store the same case schema when each job arrives.
Use a shared batch tag and an idempotency token per URL. That makes a retry safe and gives you a way to reconcile a batch that partly failed. Crucially, do not mark a candidate as disappeared after a failed collection run. Only a successful, complete run is evidence that the candidate was not observed.
Monitor source health, not just results
Monitor the public candidate-link region once a day to learn when the search set changes:
client.monitors.create(
name="google-play-examplecloud-public-search",
cron="0 7 * * *",
timezone="Europe/Paris",
job_type="standard",
target_url=search_url,
detection={
"enabled": True,
"mode": "selector",
"selector": "a[href^='/store/apps/details']",
"min_change_ratio": 0.01,
"cooldown_seconds": 86400,
"notify": {"webhook": True},
},
)
The first successful run establishes a baseline. A later result-list change emits schedule.change_detected; the webhook handler then invokes the full collection workflow above.
The operational rule is simple: a failed source is not a clean source. Public routes may return a 404, 403, a sign-in page, or a page whose selectors no longer match. Each outcome needs a visible unavailable state and an owner. None is evidence that threats disappeared.
Public social surfaces: use permitted routes only
The same case design applies to public social pages and documented public APIs, but each platform has different access rules and delivery behavior. For every source, document:
- the public or authorized route;
- query terms, languages, and regions;
- request cadence and the terms/robots review date;
- selectors or API fields that make up the candidate record; and
- the person paged when the source becomes unavailable.
Do not build a catch-all crawler for every mention of a brand. Start with high-value public surfaces where a harmful lookalike is likely to matter: an app store for a mobile product, public support impersonation pages for a SaaS product, and social surfaces where customers actually ask for help.
The review queue is the product
The durable output is not a pile of scraped pages. It is a review queue with source evidence, priority, ownership, and disposition:
public source change
→ candidate discovered
→ official assets excluded
→ detail-page evidence captured
→ explainable priority assigned
→ analyst: official / benign / report / investigate
→ source-specific reporting and customer communication, if needed
Feed the decision back into your allowlist and score rules. A false positive should make the next review faster; a confirmed abusive asset should retain enough evidence for the platform report and the incident record.
Start with one brand, one public source, and a daily schedule. Preserve the artifacts, show source health separately from “no results,” and require a human decision before acting against an app or account. ScrapeNest handles collection, artifacts, extraction, and change detection; the security team supplies the context and proportionate response.