diff --git a/AGENTS.md b/AGENTS.md index b4cf6736..9cda433f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -16,6 +16,7 @@ profile metadata. See `README.md` for the full feature list and install steps. (naming, validator signatures, orchestrator helpers, return values, style). - `docs/USAGE.md` — library-mode usage (calling the engine from Python). - `docs/FLAGS.md` — every CLI flag. +- `docs/CROSS_SCAN.md` — how `--cross-scan` mines scan metadata for usernames. - `docs/PATTERNS.md` — the username/email permutation pattern syntax. ## Repository layout diff --git a/README.md b/README.md index 309a9cee..b11179fc 100644 --- a/README.md +++ b/README.md @@ -50,6 +50,7 @@ The ultimate reconnaissance tool for hunting down targets using just an email or - ✅ **Modular & Extensible:** Built on a highly decoupled, modular architecture, adding new platform modules takes just a few lines of code. - ✅ **Mass Bulk Scanning:** High-throughput processing for bulk lists of usernames and emails via structured input files. - ✅ **Permutation Generator:** Wildcard-based username variation generation to catch typosquatting or alternative aliases. +- ✅ **Cross-Scan Pivoting:** Turns any scan into the next one — mines the handles, profile links and email addresses the results expose, classifies each by how strongly the source vouches for it, scans them across every module of their kind, and scores every hit so a handle collision is never mistaken for the target. - ✅ **Multi-Format Export:** Clean console output paired with structured, automated exports to **PDF**, **JSON** and **CSV** for easy pipeline integration. - ✅ **Advanced Proxy Rotation:** Built-in proxy pivoting with automated rotation and pre-scan health checks to bypass strict rate-limiting. - ✅ **Smart Auto-Update System:** Keeps your signatures and modules fresh with interactive, seamless PyPI update prompts. @@ -145,6 +146,38 @@ user-scanner -ef emails.txt # bulk email scan user-scanner -uf usernames.txt # bulk username scan ``` +### Cross-scan + +An email scan proves an account exists but rarely learns its name. `--cross-scan` +mines the usernames, profile links **and email addresses** the results expose, +then scans each against the modules for its own kind — reaching sites no single +pass can see. All four directions work off one mechanism: + +| Direction | Mines | +| --- | --- | +| `-e` → username | a handle the address's profile reports, or a link it carries | +| `-u` → username | the person's other handles, advertised on the profiles found | +| `-u` → email | an address published on a profile the username pass found | +| `-e` → email | a second address exposed by the first one's profiles | + +```bash +user-scanner -u johndoe --cross-scan # pivot from a username pass +user-scanner -e johndoe@gmail.com --cross-scan # pivot on every link +user-scanner -e johndoe@gmail.com --cross-scan --cross-links verified # only platform-verified links +user-scanner -u johndoe --cross-scan --cross-emails all # include addresses found in bio text +user-scanner -u johndoe --cross-scan --cross-emails none # never scan an extracted address +user-scanner -e johndoe@gmail.com --cross-scan --cross-sweep 0 # only sites a link named +user-scanner -e johndoe@gmail.com --cross-scan --cross-depth 2 # follow links a second hop +``` + +A common handle collides with other people, so every hit is rated `confirmed` / +`likely` / `candidate` / `conflicting` against the profiles the target's own +links confirmed. Addresses are rated the same way before being scanned, with two +sites publishing the same one outranking a single mention — and the email +modules that notify the address are skipped unless `--allow-loud`. See +[Cross-scan](docs/CROSS_SCAN.md) for the classes, confidence rules and cost +model. + ### Pattern generation See [Pattern Syntax](docs/PATTERNS.md) for more details diff --git a/docs/CROSS_SCAN.md b/docs/CROSS_SCAN.md new file mode 100644 index 00000000..b90c5c6e --- /dev/null +++ b/docs/CROSS_SCAN.md @@ -0,0 +1,358 @@ +# Cross-scan + +An email scan answers *does an account exist here*. It almost never learns the +account's **name**, so it can only ever reach the sites that expose an email +check. A username scan reaches far more sites, but needs a handle to start from. + +`--cross-scan` bridges the two: it runs the scan, mines the metadata the results +carry for usernames **and email addresses**, and scans each against the modules +for its own kind. + +``` +user-scanner -e target@example.com --cross-scan +user-scanner -u target --cross-scan +``` + +Either pass can be the source as well as the destination, so all four directions +work off one mechanism: + +| Direction | What it mines | +| --- | --- | +| `-e` → username | a handle the email's profile reports, or a link it carries | +| `-u` → username | the person's *other* handles, advertised on the profiles found | +| `-u` → email | an address published on a profile the username pass found | +| `-e` → email | a second address exposed by the first one's profiles | + +A pass's own target starts out excluded — a `-u` handle as already swept, a `-e` +address as already scanned — since that pass ran every module against it. + +--- + +## Where pivots come from + +Two shapes of metadata carry a username: + +| Shape | Example | Becomes | +| --- | --- | --- | +| A handle the site reports for the email | Gravatar `username: johndoe` | username `johndoe` | +| A link on the profile | `https://github.com/johndoe` | username `johndoe`, site `github` | + +Links resolve to a `(site, username)` pair through a route table in +`user_scanner/core/pivots.py` — hosts, path shapes (`/in/{user}`, +`/users/{id}/{user}`, `/@{user}`) and `{user}.host` subdomains. A link to a root +domain with no path (`https://johndoe.com/`) yields the domain label as a +username with no particular site attached. + +A link whose path names a site page rather than a person +(`github.com/settings`, `youtube.com/channel/UC…`) yields nothing — an ID is not +a handle, and a reserved word is not a person. + +--- + +## Link classes + +Pivots are classified by how much the source platform vouches for them: + +| Class | Meaning | Example | +| --- | --- | --- | +| `handle` | The site itself reported this account's name for the scanned email | Gravatar's `username` | +| `verified` | The owner proved control of the far side — OAuth connection or a `rel="me"` round-trip | Gravatar's `verified_accounts` | +| `link` | Free text the owner typed into their profile | Gravatar's `links`, `websites`, `bio` | + +`--cross-links` picks which classes may be pivoted from: + +| Value | Uses | +| --- | --- | +| `all` (default) | every class | +| `verified` | `handle` + `verified` — nothing the owner could have typed | +| `none` | `handle` only | + +`verified` is the setting to reach for when a false link would be costly: anyone +can paste someone else's URL into their own bio, but they cannot complete the +platform's verification handshake for an account they do not control. + +--- + +## Email classes + +Addresses are classified the same way, by how the source presented them: + +| Class | Meaning | Example | +| --- | --- | --- | +| `field` | The site published it in its own email field for the account | GitHub's `email`, Gravatar's `emails` | +| `text` | An address read out of prose, where nothing says whose mailbox it is | an address inside a `bio` | + +`--cross-emails` picks which may be scanned: + +| Value | Uses | +| --- | --- | +| `all` | both classes | +| `verified` (default) | `field` only | +| `none` | nothing — no address is scanned | + +It defaults tighter than `--cross-links` because the cost of being wrong is not +symmetric. A stray username pivot wastes a request; a stray address puts a third +party into the report, and hands their mailbox to modules that can write to it. + +Two traps this classification exists to avoid: + +- **An email field is not a guarantee of ownership.** `verified` says the site + published the address, not that the site was right about whose it is. PyPI + fills its `email` from a package's author/maintainer metadata, so a hit there + can carry a co-maintainer's address or a mailing list. Keys that name the + third party outright (`author_email`, `maintainer_email`) are read as `text`, + but a module that folds them into `email` defeats that. +- **Some addresses reach nobody.** Role mailboxes (`noreply@`, `postmaster@`), + RFC 2606 placeholders (`@example.com`), reserved TLDs and GitHub's + `@users.noreply.github.com` relay are dropped outright. `hello@` and + `contact@` are *not* — that is how a freelancer takes mail. + +`none` means none, unlike its `--cross-links` namesake, which still yields +handle pivots. A handle is not a link; every email class is an address. + +--- + +## The sweep, and why hits are not equal + +Two very different things produce a hit: + +| | What it proves | +| --- | --- | +| **Named check** — a pivot gave the site *and* the handle (`github.com/johndoe`) | The target's own profile pointed here | +| **Sweep** — the handle tried on every other module | The handle is registered there, by *anyone* | + +A common handle collides. One sweep of a plausible handle turned up five +different people alongside the real owner — so a sweep hit is a lead, not an +identification. + +`--cross-sweep 0` turns the sweep off and runs only the named checks. Far fewer +accounts, zero collisions. + +### Usernames and addresses share the budget + +`--cross-sweep` counts *targets*, not usernames: sweeping either kind costs one +full pass over its scan type (227 username modules, 153 email ones). Half the +budget is offered to addresses, rounded down, and whatever one kind cannot use +falls to the other: + +| Budget | Usernames available | Addresses available | Spent on | +| --- | --- | --- | --- | +| 3 | 5 | 2 | 2 usernames, 1 address | +| 3 | 0 | 2 | 2 addresses | +| 3 | 5 | 0 | 3 usernames | +| 1 | 2 | 2 | 1 username | + +A budget of 1 still goes to a username, which is what it did before addresses +existed. `--cross-sweep 0` leaves only named checks, so no address is scanned — +an address has no named-site equivalent to fall back on. + +--- + +## Confidence + +Every hit is rated, and the rating is written to `extra.confidence`: + +| Rating | Meaning | +| --- | --- | +| `confirmed` | A pivot named this exact site and handle | +| `likely` | Metadata matches the confirmed profiles | +| `candidate` | The handle is registered; nothing ties it to the target | +| `conflicting` | Metadata names someone else | + +`likely` and `conflicting` are decided against **anchors** — the names, personal +domains, e-mail addresses, profile URLs and confirmed *accounts* harvested from +the `confirmed` hits. A hit that echoes an anchor is promoted; a hit whose name +field reads as a different person's name is demoted. + +The strongest of those signals is a link **to a confirmed account**. If X is +confirmed and a swept Twitch profile links that same X account, the Twitch +account is `likely` — someone else holding the handle would not advertise the +target's Twitter. Matching is on the resolved `(site, handle)` pair rather than +the URL text, so a renamed host or a different casing still lands: +`twitter.com/JohnDoe2` and `x.com/johndoe2` both resolve to `("x", "johndoe2")`. + +Links are read from **every** field a module emits, not a fixed list of text +keys — sites split them across `bio`, `website`, `showcased_links`, or one key +per platform, and a Linktree that lists three confirmed accounts should not go +unrated because its field happens to be named something new. + +Two rules keep the demotion honest. A name that merely restates the handle +(`john.d.oe`, `JohnDoe`) is an echo of the search rather than evidence, so it is +ignored. And a field that packs a descriptor around the name (some sites +render one as `Other Person, 44, male`) is read up to its first comma, so the +rest is not mistaken for a mismatch. + +Scoring runs after the pass finishes, because the anchors come from that same +pass's confirmed hits. Ratings therefore appear in the export and the closing +summary, not on the per-result lines as they stream past. + +### Addresses are rated before they are scanned + +An address is rated on how independently it was reported, and the accounts it +finds inherit that rating — an account is only as well tied to the target as the +address that led to it: + +| Rating | Earned by | +| --- | --- | +| `confirmed` | two or more sites published it in their own email field | +| `likely` | one site published it in an email field, or it sits on a domain the target links to | +| `candidate` | prose only, with nothing tying it back | + +Independent agreement is the strongest signal available without sending mail, so +it outranks a single site saying it once. `conflicting` is never used: an +address carries no name to disagree with, and inferring a mismatch from the +local part would mislabel every shared mailbox. + +The rating deliberately ignores the anchor *emails* and *domains* — those are +harvested from the very profiles being rated, so consulting them would promote +every address on the strength of its own appearance. Only domains the target was +seen to **link** to count. + +### What confidence does not do + +- **A `candidate` is not a negative.** Most hits land there simply because the + site exposes no metadata to judge — `Roblox`, `Scratch` and `Px500` return a + handle and little else. +- **Location is not used.** "Brazil" fits millions of people, and people move, + so a location match would promote hits it cannot justify. +- **Nothing is dropped.** Every hit reaches the export whatever its rating. + +--- + +## Scope + +`-m` and `-c` narrow the cross-scan exactly as they narrow the first pass, so a +restricted run stays restricted: + +``` +-u johndoe -m gravatar --cross-scan # 1 module in the first pass, 1 in the sweep +-u johndoe -c dev --cross-scan # 44 dev modules in both +``` + +Both the sweep and the named checks honour it, so a pivot naming a site outside +the restriction is not checked either. Names are re-resolved against `user_scan`, +because an email run's `-m` names *email* modules while the sweep needs the +username module of the same site. + +Addresses resolve the same name against `email_scan`, so `-m github` narrows the +sweep to `user_scan/dev/github.py` and any address scan to +`email_scan/dev/github.py`. Only a restriction that names **neither** a username +nor an email module leaves nothing to cross-scan, and the run says so. + +### Loud modules are skipped, not prompted + +23 email modules notify the address they are given — a password reset or a +verification mail. In a first pass that address is the one you typed, so +`--allow-loud` and a per-module prompt are the right bar. In a cross-scan it +came off somebody else's profile, so those modules are dropped without asking. +`--allow-loud` puts them back for a caller who has accepted that. + +--- + +## Depth: following a chain of links + +`--cross-depth N` runs N rounds. Each round pivots off the accounts the previous +one found, so a handle that only appears deep in a chain is still reached: + +``` +Gravatar --verified--> Dev.to --website--> somebrand.com -> handle "somebrand" +``` + +Round 1 never sees that handle: Gravatar does not link Dev.to, and `somebrand` +appears nowhere in the email results. Only a second round reaches it. + +Breadth and depth are separate axes, so they combine: + +| | `--cross-sweep N` (default 3) | `--cross-sweep 0` | +| --- | --- | --- | +| `--cross-depth 1` | every module × the top handles | only the sites links named | +| `--cross-depth 2` | the above, plus handles found one hop deeper | follow links two hops, still never guessing | + +`--cross-sweep 0 --cross-depth 2` is the cheap, high-precision mode: it walks +the link graph without ever trying a handle on a site nothing pointed at, so it +cannot produce a collision. + +Two rules keep extra rounds from wandering: + +- **Nothing is scanned twice.** A swept username has already had every module run + against it, so it is never revisited; a named site+handle pair is retired once + checked. +- **A `conflicting` account is not followed.** Its metadata names someone else, + so its links lead into a stranger's footprint rather than the target's. + +--- + +## Cost + +A sweep runs **every** module of its kind, so each swept username costs roughly +one full `-u` scan and each scanned address roughly one full `-e` scan. +`--cross-sweep` is that shared budget: it caps how many targets get the +treatment (default 3) **across all rounds and both kinds**, and `0` turns +sweeping off altogether. Raising `--cross-depth` never multiplies the bill on +its own — a deeper run with the default budget spends it in round 1 and reaches +later rounds with named checks only. Raise both together. + +Usernames are ranked `handle` → `verified` → `link`, then by how many pivots +mention them. Those past the budget are named in the output rather than dropped +silently, and any pivot that named a specific site is still checked against that +one site — cheap, and it keeps a capped run from missing a confirmed link. + +Ranking is by how well a handle is vouched for, not by how useful it looks, so +an opaque platform ID that arrived through a verified link (Spotify hands out +`21jxv335g4w6dikpyyhtlbybq`) can outrank a real handle and spend a sweep on a +string no other site will ever have. Watch the pivot table and raise the cap, or +drop to `--cross-sweep 0`, when a run is full of them. + +--- + +## Reading the output + +Cross-scan hits carry `pivot_source` (why the target was scanned) and +`confidence` (how well the account is tied to the target). An account reached +through an address records which profile published it: + +```json +{ + "status": "Found", + "username": "john@acme.dev", + "site_name": "Spotify", + "is_email": true, + "extra": { + "pivot_source": "address from Github (email), Gravatar (emails)", + "confidence": "confirmed" + } +} +``` + +and one reached through a handle records the pivot class: + +```json +{ + "status": "Found", + "username": "johndoe", + "site_name": "Github", + "extra": { + "pivot_source": "verified from Gravatar (verified_accounts)", + "confidence": "confirmed" + } +} +``` + +The closing summary counts each rating, names the `confirmed`, `likely` and +`conflicting` hits, and lists the sites the second pass reached that the first +pass never did. + +--- + +## Limits + +- A username pass (`-u` / `-uf`) can be cross-scanned too. Its own target starts + out marked as swept, since that pass already ran every module against it and + most sites report the handle straight back as a pivot. +- Link shorteners are dead ends. `t.co/abc123` yields no pivot, and the redirect + is never followed, so whatever it points at stays invisible at any depth. +- A pivot is a lead, not proof of identity. A link on a profile says the profile + owner pointed at that account, and `verified` says the platform checked it — + neither says the two accounts belong to the same person in every case. +- Confidence is a triage aid, not a verdict. It reads only the metadata a module + happened to extract, so it cannot rate a site that exposes none. diff --git a/docs/FLAGS.md b/docs/FLAGS.md index 280d77ed..a94897ca 100644 --- a/docs/FLAGS.md +++ b/docs/FLAGS.md @@ -9,12 +9,17 @@ | `--allow-loud` | Enable scanning sites that may send emails/notifications | | `--no-nsfw` | Disable NSFW site scanning | | `--hudson, --hudson-scan` | Check for infostealer intelligence using Hudson Rock's API | -| `-c, --category CATEGORY` | Scan all platforms in a specific category (comma-separated for multiple) | +| `--cross-scan` | After the scan, follow the usernames, links and email addresses its results expose and scan those too (see [CROSS_SCAN.md](CROSS_SCAN.md)) | +| `--cross-links {all,verified,none}` | Which links a cross-scan may pivot from (default: `all`) | +| `--cross-emails {all,verified,none}` | Which addresses a cross-scan may scan as emails: `all` includes ones scraped from bio text, `verified` only ones a site published in its own email field, `none` scans none. Loud email modules are skipped unless `--allow-loud` (default: `verified`) | +| `--cross-depth N` | Rounds of link-following; each round pivots off the accounts the previous one found (default: 1) | +| `--cross-sweep N` | Targets — usernames and addresses together — swept against every module of their kind, across all rounds; `0` disables sweeping (default: 3) | +| `-c, --category CATEGORY` | Scan all platforms in a specific category (comma-separated for multiple); also narrows `--cross-scan` | | `-lu, --list-user` | List all available modules for username scanning | | `-le, --list-email` | List all available modules for email scanning | | `-v, --verbose` | Enable verbose output to show urls of the websites | | `--all` | Show all results including Not Found/Not Registered/Error/Skipped | -| `-m, --module MODULE` | Scan a specific module (comma-separated for multiple) | +| `-m, --module MODULE` | Scan a specific module (comma-separated for multiple); also narrows `--cross-scan` | | `-p, --permute PERMUTE` | Generate username permutations using a pattern/suffix | | `-P, --proxy-file FILE` | Use proxies from file (one per line) | | `--validate-proxies` | Validate proxies before scanning (tests against google.com) | diff --git a/tests/test_confidence.py b/tests/test_confidence.py new file mode 100644 index 00000000..7d408d20 --- /dev/null +++ b/tests/test_confidence.py @@ -0,0 +1,213 @@ +from user_scanner.core.confidence import Confidence, build_anchors, rank_emails, score +from user_scanner.core.pivots import extract_email_pivots, select_email_pivots +from user_scanner.core.result import Result + +def hit(site_name, username="johndoe", **extra): + return Result.taken(extra=extra).update(site_name=site_name, username=username) + + +def anchors_from(*confirmed, emails=("johndoe@gmail.com",)): + return build_anchors( + confirmed=confirmed, + emails=emails, + urls=("https://github.com/johndoe", "https://stackoverflow.com/users/12345/johndoe"), + ) + + +CONFIRMED = hit( + "Github", + name="Johnathan Doe", + website="https://johndoe.com", + links="https://johndoe.com, https://twitter.com/johndoe2", +) + + +def test_anchors_take_names_domains_and_emails_from_confirmed_hits(): + anchors = anchors_from(CONFIRMED) + + assert "johnathandoe" in anchors.names + assert "johndoe.com" in anchors.domains + assert "johndoe@gmail.com" in anchors.emails + + +def test_anchors_ignore_platform_hosts(): + anchors = anchors_from(CONFIRMED) + + assert "twitter.com" not in anchors.domains + assert "github.com" not in anchors.domains + + +def test_anchors_ignore_a_name_that_only_repeats_the_handle(): + anchors = anchors_from(hit("Github", name="JohnDoe")) + + assert "johndoe" not in anchors.names + + +def test_another_accounts_handle_does_not_suppress_a_real_name(): + anchors = anchors_from( + hit("Linkedin", name="Johnathan Doe"), + hit("Youtube", username="JohnathanDoe"), + ) + + assert "johnathandoe" in anchors.names + + +def test_a_named_site_is_confirmed(): + rating = score(hit("Github"), anchors_from(CONFIRMED), confirmed=True) + + assert rating is Confidence.CONFIRMED + + +def test_a_matching_name_is_likely(): + rating = score(hit("Behance", name="Johnathan Doe"), anchors_from(CONFIRMED)) + + assert rating is Confidence.LIKELY + + +def test_a_personal_domain_in_the_bio_is_likely(): + rating = score(hit("Bluesky", bio="Dev - https://johndoe.com/"), anchors_from(CONFIRMED)) + + assert rating is Confidence.LIKELY + + +def test_a_confirmed_profile_url_in_the_bio_is_likely(): + rating = score( + hit("Instagram", bio="http://stackoverflow.com/users/12345/johndoe"), + anchors_from(CONFIRMED), + ) + + assert rating is Confidence.LIKELY + + +def test_linking_a_confirmed_account_is_likely(): + anchors = anchors_from(hit("X (Twitter)", username="JohnDoe2")) + twitch = hit("Twitch", twitter="https://twitter.com/JohnDoe2") + + assert score(twitch, anchors) is Confidence.LIKELY + + +def test_a_link_matches_a_confirmed_account_across_a_renamed_host(): + anchors = anchors_from(hit("X (Twitter)", username="JohnDoe2")) + + for url in ("https://x.com/johndoe2", "https://twitter.com/JohnDoe2/"): + assert score(hit("Linktree", showcased_links=url), anchors) is Confidence.LIKELY + + +def test_a_link_to_an_unconfirmed_account_is_not_corroboration(): + anchors = anchors_from(hit("X (Twitter)", username="JohnDoe2")) + other = hit("Linktree", showcased_links="https://x.com/somebodyelse") + + assert score(other, anchors) is Confidence.CANDIDATE + + +def test_the_scanned_email_in_the_bio_is_likely(): + rating = score(hit("Hackernews", bio="Reach me at johndoe@gmail.com"), anchors_from(CONFIRMED)) + + assert rating is Confidence.LIKELY + + +def test_a_different_persons_name_conflicts(): + rating = score(hit("Chess.com", name="Other Person"), anchors_from(CONFIRMED)) + + assert rating is Confidence.CONFLICTING + + +def test_a_descriptor_field_conflicts_on_its_name_part_only(): + rating = score( + hit("Somesite", i_am="Other Person, 44, male"), anchors_from(CONFIRMED) + ) + + assert rating is Confidence.CONFLICTING + + +def test_a_descriptor_carrying_no_surname_does_not_conflict(): + rating = score(hit("Somesite", i_am="Someone, 44, male"), anchors_from(CONFIRMED)) + + assert rating is Confidence.CANDIDATE + + +def test_a_name_that_only_renders_the_handle_stays_a_candidate(): + rating = score(hit("Picsart", name="john.d.oe"), anchors_from(CONFIRMED)) + + assert rating is Confidence.CANDIDATE + + +def test_a_hit_with_no_metadata_stays_a_candidate(): + rating = score(hit("Roblox"), anchors_from(CONFIRMED)) + + assert rating is Confidence.CANDIDATE + + +def test_nothing_conflicts_when_there_is_no_confirmed_account_to_conflict_with(): + rating = score(hit("Chess.com", name="Other Person"), anchors_from(emails=())) + + assert rating is Confidence.CANDIDATE + + +def emails_from(*results, mode="all"): + return select_email_pivots(extract_email_pivots(results), mode) + + +def rate(*results, mode="all"): + pivots = emails_from(*results, mode=mode) + ranked = rank_emails(pivots, build_anchors(confirmed=results)) + return {entry.email: entry.confidence for entry in ranked} + + +def test_two_sites_publishing_one_address_confirm_it(): + """Independent agreement is the strongest tie available without mailing it.""" + ratings = rate( + hit("Github", email="john@acme.dev"), + hit("Gravatar", emails="john@acme.dev"), + ) + + assert ratings == {"john@acme.dev": Confidence.CONFIRMED} + + +def test_a_single_email_field_is_only_likely(): + assert rate(hit("Github", email="john@acme.dev")) == {"john@acme.dev": Confidence.LIKELY} + + +def test_an_address_scraped_from_prose_is_a_candidate(): + assert rate(hit("Reddit", bio="mail loose@random.net")) == { + "loose@random.net": Confidence.CANDIDATE + } + + +def test_an_address_on_a_linked_domain_is_likely(): + ratings = rate(hit("Github", website="https://acme.dev", bio="me@acme.dev, other@elsewhere.io")) + + assert ratings["me@acme.dev"] is Confidence.LIKELY + assert ratings["other@elsewhere.io"] is Confidence.CANDIDATE + + +def test_an_address_does_not_vouch_for_itself(): + """build_anchors mines these same profiles for addresses and their domains, + so rating against either would promote every hit on its own appearance.""" + ratings = rate(hit("Reddit", bio="only@nowhere-else.example.dev")) + + assert ratings == {"only@nowhere-else.example.dev": Confidence.CANDIDATE} + + +def test_best_tied_addresses_come_first(): + ranked = rank_emails( + emails_from( + hit("Reddit", bio="loose@random.net"), + hit("Github", email="john@acme.dev"), + hit("Gravatar", emails="john@acme.dev"), + ), + build_anchors(confirmed=()), + ) + + assert [e.email for e in ranked] == ["john@acme.dev", "loose@random.net"] + + +def test_an_address_is_never_rated_conflicting(): + """An address carries no name to disagree with, so inventing a mismatch + from the local part would mislabel every shared mailbox.""" + ratings = rate( + hit("Github", name="Johnathan Doe", email="john@acme.dev"), + hit("Reddit", name="Other Person", bio="someone@elsewhere.io"), + ) + + assert Confidence.CONFLICTING not in ratings.values() diff --git a/tests/test_cross_scan.py b/tests/test_cross_scan.py new file mode 100644 index 00000000..224d3539 --- /dev/null +++ b/tests/test_cross_scan.py @@ -0,0 +1,193 @@ +import pytest + +from user_scanner.core.cross_scan import ( + CrossScanConfig, + _already_scanned, + _already_swept, + _email_scope, + _fresh_emails, + _followable, + _fresh_pivots, + _named_targets, + _scope, + _split_budget, +) +from user_scanner.core.helpers import ScanConfig, get_site_name, is_loud +from user_scanner.core.pivots import PivotKind +from user_scanner.core.result import Result + + +def source(**extra): + return [Result.taken(extra=extra).update(site_name="Gravatar", is_email=True)] + + +GRAVATAR = source(username="johndoe", links="https://github.com/johndoe") + + +def test_fresh_pivots_returns_everything_on_the_first_round(): + pivots = _fresh_pivots(GRAVATAR, "all", swept=set(), checked=set()) + + assert {(p.site, p.username) for p in pivots} == {("gravatar", "johndoe"), ("github", "johndoe")} + + +def test_a_swept_username_is_never_revisited(): + pivots = _fresh_pivots(GRAVATAR, "all", swept={"johndoe"}, checked=set()) + + assert pivots == [] + + +def test_a_checked_pair_is_not_rechecked(): + pivots = _fresh_pivots(GRAVATAR, "all", swept=set(), checked={("github", "johndoe")}) + + assert {p.site for p in pivots} == {"gravatar"} + + +def test_a_siteless_pivot_survives_until_its_username_is_swept(): + pivots = _fresh_pivots(source(bio="https://johndoe.com/"), "all", set(), {("x", "johndoe")}) + + assert [(p.site, p.username, p.kind) for p in pivots] == [(None, "johndoe", PivotKind.LINK)] + + +def test_only_non_conflicting_hits_are_followed(): + hits = [ + Result.taken(extra={"confidence": "confirmed"}).update(site_name="Github"), + Result.taken(extra={"confidence": "candidate"}).update(site_name="Roblox"), + Result.taken(extra={"confidence": "conflicting"}).update(site_name="Chess.com"), + Result.available().update(site_name="Steam"), + ] + + assert [r.site_name for r in _followable(hits)] == ["Github", "Roblox"] + + +def test_the_same_account_linked_twice_is_checked_once(): + """Two profiles linking one account — in any casing — is one request.""" + pivots = _fresh_pivots( + [ + Result.taken(extra={"verified_accounts": "X: https://x.com/JohnDoe2 (verified)"}) + .update(site_name="Gravatar", is_email=True), + Result.taken(extra={"links": "https://twitter.com/johndoe2"}) + .update(site_name="Linktree", is_email=True), + ], + "all", + swept=set(), + checked=set(), + ) + targets = _named_targets(pivots, swept=set(), checked=set(), configs=ScanConfig()) + + assert len(pivots) == 2 + assert [m.__name__ for mods in targets.values() for m in mods] == ["x"] + # The best-vouched pivot supplies the casing that gets scanned. + assert list(targets) == ["JohnDoe2"] + + +def test_a_username_pass_does_not_rescan_its_own_target(): + """A -u pass is already a sweep of its own handle, and sites echo that handle + back as a pivot, so it must start out marked as swept.""" + prior = [ + Result.taken(extra={"username": "JohnDoe"}).update(site_name="Chess.com", username="JohnDoe") + ] + + assert _already_swept(prior) == {"johndoe"} + assert _fresh_pivots(prior, "all", _already_swept(prior), set()) == [] + + +def test_an_email_pass_seeds_nothing(): + prior = [ + Result.taken(extra={"username": "johndoe"}).update( + site_name="Gravatar", username="johndoe@gmail.com", is_email=True + ) + ] + + assert _already_swept(prior) == set() + assert [p.username for p in _fresh_pivots(prior, "all", set(), set())] == ["johndoe"] + + +def test_an_unrestricted_run_has_no_scope(): + assert _scope(CrossScanConfig(), ScanConfig()) is None + + +def test_a_module_restriction_resolves_against_user_scan(): + """-m names email modules on an email run, so the sweep must re-resolve the + same site names against user_scan.""" + scope = _scope(CrossScanConfig(modules=("github", "chess.com")), ScanConfig()) + + assert sorted(m.__name__ for m in scope) == ["chess_com", "github"] + + +def test_a_category_restriction_resolves_to_that_folder(): + scope = _scope(CrossScanConfig(categories=("donation",)), ScanConfig()) + + assert scope and all("donation" in str(m.__file__) for m in scope) + + +def test_named_checks_stay_inside_the_scope(): + """A pivot naming a site outside -m/-c must not be checked either.""" + pivots = _fresh_pivots( + source(verified_accounts="GitHub: https://github.com/johndoe (verified), " + "LinkedIn: https://www.linkedin.com/in/johndoe (verified)"), + "all", + swept=set(), + checked=set(), + ) + scope = _scope(CrossScanConfig(modules=("github",)), ScanConfig()) + targets = _named_targets(pivots, set(), set(), ScanConfig(), scope) + + assert [m.__name__ for mods in targets.values() for m in mods] == ["github"] + + +def user_hit(site_name, **extra): + return Result.taken(extra=extra).update(site_name=site_name, username="johndoe") + + +def test_an_email_pass_does_not_rescan_its_own_target(): + prior = [Result.taken().update(site_name="Spotify", username="John@Acme.dev", is_email=True)] + + assert _already_scanned(prior) == {"john@acme.dev"} + assert _fresh_emails(prior, "all", _already_scanned(prior)) == [] + + +def test_a_username_pass_seeds_no_scanned_address(): + assert _already_scanned([user_hit("Github", email="john@acme.dev")]) == set() + + +def test_verified_is_the_default_and_drops_prose_addresses(): + source = [user_hit("Github", email="john@acme.dev", bio="also loose@random.net")] + + assert [e.email for e in _fresh_emails(source, "verified", set())] == ["john@acme.dev"] + assert len(_fresh_emails(source, "all", set())) == 2 + assert _fresh_emails(source, "none", set()) == [] + + +@pytest.mark.parametrize( + "budget,usernames,emails,expected", + [ + (3, 5, 2, (2, 1)), + (3, 0, 2, (0, 2)), # nothing to sweep, addresses take the lot + (3, 5, 0, (3, 0)), # no addresses, usernames keep the pre-existing budget + (1, 2, 2, (1, 0)), # a budget of 1 still sweeps a username first + (2, 1, 3, (1, 1)), + (0, 5, 5, (0, 0)), + ], +) +def test_neither_target_kind_starves_the_other(budget, usernames, emails, expected): + assert _split_budget(budget, usernames, emails) == expected + + +def test_loud_email_modules_are_skipped_rather_than_prompted(): + """The addresses reaching a cross-scan came off somebody else's profile, so + mailing them is not a decision this pass gets to make.""" + quiet = _email_scope(CrossScanConfig(), ScanConfig()) + loud_names = {m.__name__ for m in quiet if is_loud(get_site_name(m), is_email=True)} + + assert loud_names == set() + assert _email_scope(CrossScanConfig(), ScanConfig(allow_loud=True)) is None + + +def test_a_module_restriction_resolves_against_email_scan(): + scope = _email_scope(CrossScanConfig(modules=("github",)), ScanConfig()) + + assert [m.__name__ for m in scope] == ["github"] + + +def test_emails_none_loads_no_module_at_all(): + assert _email_scope(CrossScanConfig(emails="none"), ScanConfig()) == [] diff --git a/tests/test_pivots.py b/tests/test_pivots.py new file mode 100644 index 00000000..8aa802fc --- /dev/null +++ b/tests/test_pivots.py @@ -0,0 +1,293 @@ +from pathlib import Path + +import pytest + +from user_scanner.core.pivots import ( + EmailKind, + _HOST_ROUTES, + _SUBDOMAIN_ROUTES, + Pivot, + PivotKind, + extract_email_pivots, + extract_pivots, + is_platform_host, + rank_usernames, + resolve_url, + select_email_pivots, + select_pivots, +) +from user_scanner.core.result import Result + +USER_SCAN_ROOT = Path(__file__).resolve().parent.parent / "user_scanner" / "user_scan" + + +def make_result(site_name="Gravatar", extra=None, found=True, **kwargs): + factory = Result.taken if found else Result.available + return factory(extra=extra or {}, **kwargs).update(site_name=site_name, is_email=True) + + +@pytest.mark.parametrize( + "url,expected", + [ + ("https://github.com/johndoe", ("github", "johndoe")), + ("https://www.linkedin.com/in/johndoe/", ("linkedin", "johndoe")), + ("https://br.linkedin.com/in/johndoe", ("linkedin", "johndoe")), + ("https://x.com/JohnDoe2", ("x", "JohnDoe2")), + ("https://twitter.com/JohnDoe2", ("x", "JohnDoe2")), + ("https://stackoverflow.com/users/12345/johndoe", ("stackoverflow", "johndoe")), + ("https://www.youtube.com/@johndoe", ("youtube", "johndoe")), + ("https://mastodon.social/@johndoe", ("mastodon", "johndoe")), + ("https://johndoe.tumblr.com/", ("tumblr", "johndoe")), + ("https://johndoe.github.io", ("github", "johndoe")), + ("https://bsky.app/profile/johndoe.bsky.social", ("bluesky", "johndoe")), + ("https://www.reddit.com/user/johndoe/", ("reddit", "johndoe")), + ], +) +def test_resolve_url_reads_the_handle(url, expected): + assert resolve_url(url) == expected + + +@pytest.mark.parametrize( + "url", + [ + "https://www.youtube.com/channel/UCMandQh49QaAH2ZfHWZdEFA", + "https://open.spotify.com/user/21jxv335g4w6dikpyyhtlbybq", + "https://github.com/settings/profile", + "https://x.com/i/flow/login", + "https://github.com/", + "ftp://github.com/johndoe", + "not a url", + ], +) +def test_resolve_url_rejects_non_profiles(url): + assert resolve_url(url) == (None, None) + + +def test_a_platform_with_no_portable_handle_is_still_a_platform(): + """A routed host with no path pattern yields no pivot, but must not be + mistaken for the target's own domain.""" + assert resolve_url("https://open.spotify.com/") == (None, None) + assert is_platform_host("open.spotify.com") + + +def test_resolve_url_reads_a_personal_domain_root_only(): + assert resolve_url("https://johndoe.com/") == (None, "johndoe") + assert resolve_url("https://johndoe.com/contact") == (None, None) + + +def test_verified_accounts_outrank_owner_entered_links(): + result = make_result( + extra={ + "username": "johndoe", + "verified_accounts": "GitHub: https://github.com/johndoe (verified)", + "links": "Mine: https://x.com/JohnDoe2", + } + ) + + kinds = {(p.site, p.kind) for p in extract_pivots([result])} + + assert ("gravatar", PivotKind.HANDLE) in kinds + assert ("github", PivotKind.VERIFIED) in kinds + assert ("x", PivotKind.LINK) in kinds + + +def test_a_verified_suffix_marks_a_link_verified_under_any_key(): + result = make_result(extra={"links": "GitHub: https://github.com/johndoe (verified)"}) + + assert extract_pivots([result])[0].kind is PivotKind.VERIFIED + + +def test_misses_are_not_mined(): + result = make_result(extra={"username": "johndoe"}, found=False) + + assert extract_pivots([result]) == [] + + +def test_avatar_urls_are_not_pivots(): + result = make_result(extra={"avatar_url": "https://gravatar.com/avatar/abc123"}) + + assert extract_pivots([result]) == [] + + +def test_a_sites_own_homepage_is_not_a_username(): + result = make_result(site_name="Adobe", extra={"homepage": "https://adobe.com/"}) + + assert extract_pivots([result]) == [] + + +def test_a_platform_named_key_holding_a_bare_handle_is_a_pivot(): + result = make_result(site_name="Kick", extra={"twitter": "BrunoLM7"}) + + (pivot,) = extract_pivots([result]) + assert (pivot.username, pivot.site, pivot.kind) == ("BrunoLM7", "x", PivotKind.LINK) + + +@pytest.mark.parametrize( + "key, expected_site", + [ + ("twitter", "x"), + ("twitter_handle", "x"), + ("twitter_username", "x"), + ("instagram", "instagram"), + ("youtube", "youtube"), + ], +) +def test_platform_keys_are_read_through_their_suffixes(key, expected_site): + result = make_result(site_name="Kick", extra={key: "someone"}) + + (pivot,) = extract_pivots([result]) + assert pivot.site == expected_site + + +def test_an_identifier_field_is_not_read_as_a_handle(): + """``_id`` is not a handle suffix, so YouTube's channel id stays out.""" + result = make_result( + site_name="Youtube", extra={"youtube_channel_id": "UCMandQh49QaAH2ZfHWZdEFA"} + ) + + assert extract_pivots([result]) == [] + + +def test_a_discord_field_holds_an_invite_not_a_handle(): + """Kick stores a server invite code under ``discord``, which names no account.""" + result = make_result(site_name="Kick", extra={"discord": "nAZEkUNWPt"}) + + assert extract_pivots([result]) == [] + + +def test_a_platform_key_holding_a_url_still_goes_through_link_extraction(): + result = make_result(site_name="Npmjs", extra={"github": "https://github.com/brunolm"}) + + (pivot,) = extract_pivots([result]) + assert (pivot.username, pivot.site) == ("brunolm", "github") + + +def test_select_pivots_filters_by_link_class(): + pivots = [ + Pivot("a", PivotKind.HANDLE, "Gravatar", "username"), + Pivot("b", PivotKind.VERIFIED, "Gravatar", "verified_accounts"), + Pivot("c", PivotKind.LINK, "Gravatar", "links"), + ] + + assert [p.username for p in select_pivots(pivots, "all")] == ["a", "b", "c"] + assert [p.username for p in select_pivots(pivots, "verified")] == ["a", "b"] + assert [p.username for p in select_pivots(pivots, "none")] == ["a"] + + +def test_rank_usernames_prefers_the_best_vouched_casing(): + pivots = [ + Pivot("johndoe2", PivotKind.LINK, "Gravatar", "links"), + Pivot("JohnDoe2", PivotKind.VERIFIED, "Gravatar", "verified_accounts"), + Pivot("other", PivotKind.VERIFIED, "Gravatar", "verified_accounts"), + ] + + assert rank_usernames(pivots) == ["JohnDoe2", "other"] + + +@pytest.mark.parametrize( + "module", sorted({m for _, m, _ in _HOST_ROUTES} | {m for _, m in _SUBDOMAIN_ROUTES}) +) +def test_every_route_names_a_live_user_scan_module(module): + assert list(USER_SCAN_ROOT.glob(f"*/{module}.py")), f"no user_scan module named {module}" + + +def user_result(site_name, **extra): + return Result.taken(extra=extra).update(site_name=site_name, username="johndoe") + + +def test_a_dedicated_email_field_outranks_one_scraped_from_prose(): + pivots = extract_email_pivots( + [user_result("GitHub", email="john@acme.dev", bio="or try other@acme.dev")] + ) + + assert [(p.email, p.kind) for p in pivots] == [ + ("john@acme.dev", EmailKind.FIELD), + ("other@acme.dev", EmailKind.TEXT), + ] + + +def test_package_metadata_is_not_the_account_holders_address(): + """author_email/maintainer_email name whoever published a release, so they + stay out of the trusted tier that --cross-emails verified keeps.""" + pivots = extract_email_pivots([user_result("PyPI", author_email="maint@pkg.org")]) + + assert [p.kind for p in pivots] == [EmailKind.TEXT] + assert select_email_pivots(pivots, "verified") == [] + + +def test_the_same_address_from_two_sites_is_kept_once_per_site(): + """Frequency is the ranking signal, so per-site pivots must survive dedupe.""" + pivots = extract_email_pivots( + [user_result("GitHub", email="john@acme.dev"), user_result("Gravatar", emails="john@acme.dev")] + ) + + assert [p.source_site for p in pivots] == ["GitHub", "Gravatar"] + + +def test_casing_is_normalised_so_one_mailbox_is_one_target(): + pivots = extract_email_pivots([user_result("GitHub", email="John.Doe@Acme.DEV")]) + + assert [p.email for p in pivots] == ["john.doe@acme.dev"] + + +@pytest.mark.parametrize( + "address", + [ + "noreply@acme.dev", + "postmaster@acme.dev", + "12345+johndoe@users.noreply.github.com", + "someone@example.com", + "someone@yourdomain.com", + "someone@acme.test", + "not-an-address", + ], +) +def test_addresses_that_reach_nobody_are_dropped(address): + assert extract_email_pivots([user_result("GitHub", email=address)]) == [] + + +def test_a_role_lookalike_that_is_a_real_mailbox_survives(): + """hello@ and contact@ are how freelancers take mail, so they stay in.""" + pivots = extract_email_pivots([user_result("GitHub", email="hello@acme.dev")]) + + assert [p.email for p in pivots] == ["hello@acme.dev"] + + +def test_an_avatar_url_is_never_mined_for_an_address(): + assert extract_email_pivots([user_result("GitHub", avatar_url="https://x.dev/a@b.png")]) == [] + + +def test_none_keeps_nothing_unlike_its_links_namesake(): + """--cross-links none still yields handle pivots because a handle is not a + link; every email tier is an address, so none means none.""" + pivots = extract_email_pivots([user_result("GitHub", email="john@acme.dev")]) + + assert select_email_pivots(pivots, "all") + assert select_email_pivots(pivots, "verified") + assert select_email_pivots(pivots, "none") == [] + + +def test_a_profile_link_is_not_an_address(): + """tiktok.com/@jane.doe is a legal dot-atom address whose local part is + the host and path. Links already arrive as username pivots.""" + pivots = extract_email_pivots( + [user_result("Cam4", social_links="https://www.tiktok.com/@jane.doe")] + ) + + assert pivots == [] + + +def test_a_fediverse_handle_is_not_a_mailbox(): + pivots = extract_email_pivots( + [user_result("Sourceforge", social_networks="Mastodon: @johndoe@mastodon.social")] + ) + + assert pivots == [] + + +def test_an_address_beside_a_link_still_survives(): + pivots = extract_email_pivots( + [user_result("Reddit", bio="site https://acme.dev/@notme and mail john@acme.dev")] + ) + + assert [p.email for p in pivots] == ["john@acme.dev"] diff --git a/user_scanner/__main__.py b/user_scanner/__main__.py index dc98c8d3..89fa08de 100644 --- a/user_scanner/__main__.py +++ b/user_scanner/__main__.py @@ -10,6 +10,14 @@ from user_scanner.cli.banner import print_banner from user_scanner.core import formatter +from user_scanner.core.cross_scan import ( + DEFAULT_DEPTH, + DEFAULT_SWEEP, + EMAIL_CHOICES, + LINK_CHOICES, + CrossScanConfig, + run_cross_scan, +) from user_scanner.core.email_orchestrator import ( run_email_category_batch, run_email_full_batch, @@ -50,6 +58,14 @@ MAX_PERMUTATIONS_LIMIT = 100 +def _csv_names(value) -> tuple: + """Split a repeatable, comma-separated -m/-c value into names.""" + if not value: + return () + raw = ",".join(value) if isinstance(value, list) else value + return tuple(name.strip() for name in raw.split(",") if name.strip()) + + def main(): if "--only-found" in sys.argv: print(f"{Fore.YELLOW}[!] The '--only-found' flag is deprecated and has been removed.{Style.RESET_ALL}") @@ -167,6 +183,50 @@ def main(): help="Disable NSFW site scanning", ) + parser.add_argument( + "--cross-scan", + action="store_true", + help="After the scan, follow the usernames, links and email addresses its " + "results expose and scan those too", + ) + + parser.add_argument( + "--cross-links", + choices=list(LINK_CHOICES), + default="all", + help="Which links a cross-scan may pivot from: all, verified " + "(platform-proven connections only), or none (site-reported handles only)", + ) + + parser.add_argument( + "--cross-emails", + choices=list(EMAIL_CHOICES), + default="verified", + help="Which addresses found in scan metadata a cross-scan may scan as emails: " + "all (including ones scraped from bio text), verified (only addresses a site " + "published in its own email field), or none. Loud email modules are skipped " + "unless --allow-loud (default: verified)", + ) + + parser.add_argument( + "--cross-depth", + type=int, + default=DEFAULT_DEPTH, + help="Rounds of link-following. Each round pivots off the accounts the previous " + f"one found, reaching handles only a chain of links names (default: {DEFAULT_DEPTH})", + ) + + parser.add_argument( + "--cross-sweep", + type=int, + default=DEFAULT_SWEEP, + metavar="N", + help="Targets — usernames and addresses together — a cross-scan sweeps against " + "every module of their kind, across all rounds. 0 disables sweeping, leaving only " + "the sites a pivot named — fewer accounts, but no handle collisions " + f"(default: {DEFAULT_SWEEP})", + ) + parser.add_argument("-U", "--update", action="store_true", help="Update the tool") parser.add_argument( @@ -373,6 +433,9 @@ def main(): validated_categories = [] if args.hudson_scan: + if args.cross_scan: + print(f"{R}[✘] Error: --cross-scan cannot be used with --hudson {X}") + sys.exit(1) if args.category or args.module: print(f"{R}[✘] Error: --hudson cannot be used with -m or -c {X}") print(f"{Y}[i] Use it independently{X}") @@ -473,6 +536,22 @@ def main(): if args.hudson_scan: sys.exit(0) + if args.cross_scan: + results.extend( + run_cross_scan( + results, + config, + CrossScanConfig( + links=args.cross_links, + emails=args.cross_emails, + sweep=args.cross_sweep, + depth=args.cross_depth, + modules=_csv_names(args.module), + categories=_csv_names(args.category), + ), + ) + ) + is_pdf_export = args.format == "pdf" or (args.output and args.output.lower().endswith(".pdf")) if args.output or is_pdf_export: @@ -480,11 +559,14 @@ def main(): if is_pdf_export: version_str, _ = load_local_version() + scan_type_str = "Email" if is_email else "Username" + if args.cross_scan: + scan_type_str = f"Cross-Scan ({scan_type_str})" try: pdf_bytes = formatter.into_pdf( results, target=targets_found[0] if targets_found else "Target", - scan_type="Email" if is_email else "Username", + scan_type=scan_type_str, total_modules=len(results), include_media=not args.no_pdf_media, version=version_str, diff --git a/user_scanner/core/confidence.py b/user_scanner/core/confidence.py new file mode 100644 index 00000000..dcb00fcb --- /dev/null +++ b/user_scanner/core/confidence.py @@ -0,0 +1,345 @@ +"""Scores how strongly a cross-scan hit is tied to the scanned target. + +A username sweep proves a handle is registered somewhere; it cannot prove the +account belongs to the person who was scanned. Common handles collide, and a +single sweep routinely turns up several unrelated people holding the same one +alongside its real owner, so a sweep hit is a lead until something ties it back. + +The tie is drawn from *confirmed* hits: accounts a pivot named by site and +handle, which the target's own profile pointed at. Their names, personal +domains and profile URLs become the anchors every other hit is measured against. +""" + +import re +import unicodedata +from dataclasses import dataclass +from enum import Enum +from typing import Dict, FrozenSet, Iterable, List, Optional, Set, Tuple +from urllib.parse import urlsplit + +from user_scanner.core.helpers import EMAIL_RE +from user_scanner.core.pivots import ( + EmailKind, + EmailPivot, + is_media_key, + is_platform_host, + module_stem, + resolve_url, +) +from user_scanner.core.result import Result + +# Extra keys naming the account holder. +NAME_KEYS = ( + "name", + "fullname", + "full_name", + "display_name", + "displayname", + "real_name", + "realname", + "i_am", +) + +# Bookkeeping this module writes back onto a result — never evidence about it. +_OWN_KEYS = frozenset({"confidence", "pivot_source"}) + +# Hosts that carry no identity — a shortener or a mailbox provider says nothing +# about who owns the profile linking to it. +_GENERIC_HOSTS = frozenset( + { + "amzn.to", "bit.ly", "buff.ly", "cutt.ly", "discord.gg", "docs.google.com", + "drive.google.com", "gmail.com", "goo.gl", "google.com", "hotmail.com", + "is.gd", "lnkd.in", "outlook.com", "ow.ly", "paypal.me", "rb.gy", + "t.co", "tinyurl.com", "wa.me", "yahoo.com", + } +) + +_TOKEN_RE = re.compile(r"[A-Za-z]{2,}") + + +class Confidence(Enum): + """How well a cross-scan hit is tied to the scanned target.""" + + CONFIRMED = "confirmed" + LIKELY = "likely" + CANDIDATE = "candidate" + CONFLICTING = "conflicting" + + @property + def explanation(self) -> str: + return _EXPLANATIONS[self] + + +_EXPLANATIONS = { + Confidence.CONFIRMED: "a pivot named this exact site and handle", + Confidence.LIKELY: "metadata matches the confirmed profiles", + Confidence.CANDIDATE: "handle is registered; nothing ties it to the target", + Confidence.CONFLICTING: "metadata names someone else", +} + +ORDER = ( + Confidence.CONFIRMED, + Confidence.LIKELY, + Confidence.CANDIDATE, + Confidence.CONFLICTING, +) + + +@dataclass(frozen=True) +class Anchors: + """Identity facts a hit can be measured against.""" + + names: FrozenSet[str] + domains: FrozenSet[str] + emails: FrozenSet[str] + urls: FrozenSet[str] + accounts: FrozenSet[Tuple[str, str]] + # Domains the target was seen to *link* to, without the ones inferred from + # harvested addresses. Judging an address by `domains` would be circular — + # its own domain lands there the moment it is harvested. + link_domains: FrozenSet[str] + + +@dataclass(frozen=True) +class RankedEmail: + """An address a cross-scan may follow, and how well it is tied to the target.""" + + email: str + confidence: Confidence + sources: Tuple[str, ...] + field: bool + + +def build_anchors( + confirmed: Iterable[Result], + emails: Iterable[str] = (), + urls: Iterable[str] = (), +) -> Anchors: + """Collect identity facts from the accounts already tied to the target.""" + names: Set[str] = set() + domains: Set[str] = set() + accounts: Set[Tuple[str, str]] = set() + email_set = {email.lower().strip() for email in emails if email} + url_set = {_normalize_url(url) for url in urls if url} + url_set.discard("") + + for result in confirmed: + account = _account_of(result) + if account: + accounts.add(account) + for name in _informative_names(result): + names.add(_normalize(name)) + for text in _texts(result): + email_set.update(match.group(0).lower() for match in EMAIL_RE.finditer(text)) + for url in _urls_in(text): + url_set.add(_normalize_url(url)) + host = _personal_host(url) + if host: + domains.add(host) + if result.url: + url_set.add(_normalize_url(result.url)) + + email_set.discard("") + link_domains = domains - _GENERIC_HOSTS + domains.update(email.split("@", 1)[1] for email in email_set if "@" in email) + domains -= _GENERIC_HOSTS + + names.discard("") + return Anchors( + names=frozenset(names), + domains=frozenset(domains), + emails=frozenset(email_set), + urls=frozenset(url_set), + accounts=frozenset(accounts), + link_domains=frozenset(link_domains), + ) + + +def score(result: Result, anchors: Anchors, confirmed: bool = False) -> Confidence: + """Rate one hit against the anchors.""" + if confirmed: + return Confidence.CONFIRMED + + names = _informative_names(result) + + if any(_normalize(name) in anchors.names for name in names): + return Confidence.LIKELY + + if _links_a_confirmed_account(result, anchors): + return Confidence.LIKELY + + if _echoes_anchor(result, anchors): + return Confidence.LIKELY + + if anchors.names and any(_is_person_name(name) for name in names): + return Confidence.CONFLICTING + + return Confidence.CANDIDATE + + +def rank_emails(pivots: Iterable[EmailPivot], anchors: Anchors) -> List[RankedEmail]: + """Rate every extracted address, best-tied first. + + One handle can belong to several people, so the addresses their profiles + carry are not equally likely to be the target's. Two independent sites + publishing the same address in their own email field is the strongest signal + available without sending mail, so it outranks a single site saying it once. + + ``CONFLICTING`` is never returned: an address carries no name to disagree + with, and inventing a mismatch from the local part would mislabel every + shared mailbox. + """ + by_email: Dict[str, List[EmailPivot]] = {} + for pivot in pivots: + by_email.setdefault(pivot.email, []).append(pivot) + + ranked = [ + RankedEmail( + email=email, + confidence=_rate_email(email, group, anchors), + sources=tuple(sorted({pivot.origin for pivot in group})), + field=any(pivot.kind is EmailKind.FIELD for pivot in group), + ) + for email, group in by_email.items() + ] + return sorted(ranked, key=lambda r: (ORDER.index(r.confidence), -len(r.sources), r.email)) + + +def _rate_email(email: str, group: List[EmailPivot], anchors: Anchors) -> Confidence: + # Deliberately not consulting anchors.emails: build_anchors harvests + # addresses out of the very profiles being ranked here, so matching against + # it would promote every address on a confirmed profile to CONFIRMED on the + # strength of its own appearance. + fields = {pivot.source_site for pivot in group if pivot.kind is EmailKind.FIELD} + if len(fields) >= 2: + return Confidence.CONFIRMED + if fields: + return Confidence.LIKELY + + # link_domains, not domains: the latter absorbs the domain of every address + # harvested from these same profiles, which would rate each one LIKELY on + # the strength of its own appearance. + if email.rpartition("@")[2] in anchors.link_domains: + return Confidence.LIKELY + + return Confidence.CANDIDATE + + +def _links_a_confirmed_account(result: Result, anchors: Anchors) -> bool: + """True when the profile points at an account already tied to the target. + + Matching is on the resolved (site, handle) pair rather than on the URL text, + so an old host or a different casing still lands — a profile linking + ``twitter.com/JohnDoe2`` names the same account as a confirmed + ``x.com/JohnDoe2``. + """ + own = _account_of(result) + for text in _texts(result): + for url in _urls_in(text): + site, handle = resolve_url(url) + if not site or not handle: + continue + account = (site, handle.lower()) + if account != own and account in anchors.accounts: + return True + return False + + +def _echoes_anchor(result: Result, anchors: Anchors) -> bool: + haystack = " ".join(_texts(result)).lower() + if not haystack: + return False + if any(email in haystack for email in anchors.emails): + return True + stripped = _strip_schemes(haystack) + if any(url and url in stripped for url in anchors.urls): + return True + return any(domain in stripped for domain in anchors.domains) + + +def _account_of(result: Result) -> Optional[Tuple[str, str]]: + stem = module_stem(str(result.site_name or "")) + username = str(result.username or "").lower() + return (stem, username) if stem and username else None + + +def _is_person_name(value: str) -> bool: + """True when a value reads as somebody's name rather than a label. + + Two or more word-like tokens is the bar: ``Other Person`` names a person, + ``john.d.oe`` is a rendering of the handle. + """ + return len(_TOKEN_RE.findall(value)) >= 2 + + +def _informative_names(result: Result) -> List[str]: + """The names on a result that say more than its own handle already does. + + A profile whose display name just restates the handle it was found under + (``johndoe`` → ``john.d.oe``) is echoing the search, not naming anybody. The + comparison is against *this* account's handle: another account's handle may + legitimately be the person's full name, and must not silence it here. + """ + own = _normalize(str(result.username or "")) + return [name for name in _names(result) if _normalize(name) and _normalize(name) != own] + + +def _names(result: Result) -> List[str]: + """The name each name-bearing field claims. + + Some sites pack a descriptor into the field, rendering it as + ``Other Person, 44, male`` — so only the leading segment is the + name, and reading the whole string would call every such profile a mismatch. + """ + return [ + str(result.extra[key]).split(",", 1)[0].strip() + for key in NAME_KEYS + if result.extra.get(key) + ] + + +def _texts(result: Result) -> List[str]: + """Every value that could carry a link or an address. + + Sites name these fields as they please — ``bio``, ``website``, + ``showcased_links``, or one key per platform — so anything that is not an + image or this module's own bookkeeping is read. + """ + texts = [ + str(value) + for key, value in result.extra.items() + if key not in _OWN_KEYS and not is_media_key(key) and isinstance(value, str) + ] + if result.url: + texts.append(str(result.url)) + return texts + + +def _urls_in(text: str) -> List[str]: + return re.findall(r"https?://[^\s,;\"'<>()\[\]]+", text) + + +def _personal_host(url: str) -> str: + """The host of a URL that belongs to the target rather than to a platform.""" + try: + host = (urlsplit(url).hostname or "").lower() + except ValueError: + return "" + host = host.removeprefix("www.") + if not host or host in _GENERIC_HOSTS or is_platform_host(host): + return "" + return host + + +def _normalize_url(url: str) -> str: + return _strip_schemes(url.strip().lower()).rstrip("/") + + +def _strip_schemes(value: str) -> str: + return value.replace("https://", "").replace("http://", "").replace("www.", "") + + +def _normalize(value: str) -> str: + decomposed = unicodedata.normalize("NFKD", value or "") + without_marks = "".join(ch for ch in decomposed if not unicodedata.combining(ch)) + return re.sub(r"[^a-z0-9]+", "", without_marks.lower()) diff --git a/user_scanner/core/cross_scan.py b/user_scanner/core/cross_scan.py new file mode 100644 index 00000000..7bd07d3c --- /dev/null +++ b/user_scanner/core/cross_scan.py @@ -0,0 +1,563 @@ +"""Second scan pass driven by the first pass's metadata. + +An email scan proves an account exists but rarely names it. When a profile does +expose a handle or a link to another platform, that handle can be scanned as a +username — reaching accounts no email check can see. + +Two kinds of hit come out of that, and they are not equally trustworthy: a site +a pivot named by handle, and a site where the same handle merely happens to be +taken. Every hit is scored so the difference survives into the report. +""" + +from dataclasses import dataclass +from types import ModuleType +from typing import Dict, Iterable, List, Optional, Set, Tuple + +from colorama import Fore, Style + +from user_scanner.core.confidence import ( + ORDER, + Confidence, + RankedEmail, + build_anchors, + rank_emails, + score, +) +from user_scanner.core.email_orchestrator import run_email_full_batch, run_email_module_batch +from user_scanner.core.helpers import ( + ScanConfig, + find_module, + get_site_name, + is_loud, + load_categories, + load_modules, +) +from user_scanner.core.orchestrator import run_user_full, run_user_module +from user_scanner.core.pivots import ( + Pivot, + extract_email_pivots, + extract_pivots, + module_stem, + rank_usernames, + select_email_pivots, + select_pivots, +) +from user_scanner.core.result import Result + +DEFAULT_SWEEP = 3 +DEFAULT_DEPTH = 1 +LINK_CHOICES = ("all", "verified", "none") +EMAIL_CHOICES = ("all", "verified", "none") + +_CONFIDENCE_COLOR = { + Confidence.CONFIRMED: Fore.GREEN, + Confidence.LIKELY: Fore.CYAN, + Confidence.CANDIDATE: Fore.WHITE, + Confidence.CONFLICTING: Fore.YELLOW, +} + + +@dataclass(frozen=True) +class CrossScanConfig: + links: str = "all" + # Module and category names the run was restricted to, so -m and -c narrow + # this pass exactly as they narrowed the first one. Names, not modules: an + # email run's -m names email modules, and the sweep needs the username + # module of the same site. + modules: Tuple[str, ...] = () + categories: Tuple[str, ...] = () + # Which addresses found in scan metadata may be scanned as emails. Defaults + # tighter than `links` because the cost of being wrong is higher: a stray + # username pivot wastes a request, a stray address puts a third party in the + # report and can hand them to a module that mails them. + emails: str = "verified" + # How many targets may be swept against every module, across all rounds — + # usernames and addresses draw on the same budget, since sweeping either one + # costs a full pass over that scan type. Zero turns sweeping off entirely, + # leaving only the sites a pivot named. + sweep: int = DEFAULT_SWEEP + depth: int = DEFAULT_DEPTH + + +def run_cross_scan( + results: List[Result], configs: ScanConfig, cross_configs: CrossScanConfig +) -> List[Result]: + """Mine finished results for usernames and scan them. + + Runs ``depth`` rounds: each round pivots off the accounts the previous one + found, so a handle reachable only through a chain of profile links is still + reached. Returns every round's results; the caller owns merging them into + its own list for export. + """ + print(f"\n{Fore.MAGENTA}== CROSS-SCAN =={Style.RESET_ALL}") + + scope = _scope(cross_configs, configs) + email_modules = _email_scope(cross_configs, configs) + # Either half alone is a usable pass, so this only gives up when the + # restriction leaves neither a username module nor an email module. + if scope is not None and not scope and email_modules is not None and not email_modules: + print( + f"{Fore.YELLOW}[!] The -m/-c restriction names no username or email " + f"module, so there is nothing to cross-scan.{Style.RESET_ALL}" + ) + return [] + + depth = max(1, cross_configs.depth) + budget = max(0, cross_configs.sweep) + all_pivots: List[Pivot] = [] + cross_results: List[Result] = [] + swept: Set[str] = _already_swept(results) + checked: Set[Tuple[str, str]] = set() + scanned_emails: Set[str] = _already_scanned(results) + email_ratings: Dict[str, Confidence] = {} + source: List[Result] = list(results) + + for round_number in range(1, depth + 1): + pivots = _fresh_pivots(source, cross_configs.links, swept, checked) + ranked = _fresh_emails(source, cross_configs.emails, scanned_emails) + if not pivots and not ranked: + if round_number == 1: + print( + f"{Fore.YELLOW}[!] No usernames, links or addresses to pivot " + f"from.{Style.RESET_ALL}" + ) + return [] + print(f"\n{Fore.CYAN}[i] Round {round_number}: nothing new to follow.{Style.RESET_ALL}") + break + + if depth > 1: + print(f"\n{Fore.MAGENTA}-- round {round_number} of {depth} --{Style.RESET_ALL}") + if pivots: + _print_pivots(pivots) + if ranked: + _print_emails(ranked) + all_pivots.extend(pivots) + + sweepable = 0 if scope is not None and not scope else len(rank_usernames(pivots)) + for_usernames, for_emails = _split_budget(budget, sweepable, len(ranked)) + usernames = _sweep_targets(pivots, for_usernames) + to_scan = _email_targets(ranked, for_emails, email_modules) + budget -= len(usernames) + len(to_scan) + swept.update(username.lower() for username in usernames) + scanned_emails.update(entry.email for entry in to_scan) + email_ratings.update({entry.email: entry.confidence for entry in to_scan}) + + round_results: List[Result] = [] + for username in usernames: + print( + f"\n{Fore.CYAN}[+] Sweeping every module for username: {username}{Style.RESET_ALL}" + ) + swept_results = ( + run_user_full(username, configs) + if scope is None + else run_user_module(scope, username, configs) + ) + round_results.extend(_tag(swept_results, all_pivots, username)) + + for username, modules in _named_targets(pivots, swept, checked, configs, scope).items(): + print( + f"\n{Fore.CYAN}[+] Checking {username} on its {len(modules)} linked " + f"site(s){Style.RESET_ALL}" + ) + round_results.extend( + _tag(run_user_module(modules, username, configs), all_pivots, username) + ) + + for entry in to_scan: + print(f"\n{Fore.CYAN}[+] Scanning email: {entry.email}{Style.RESET_ALL}") + email_results = ( + run_email_full_batch(entry.email, configs) + if email_modules is None + else run_email_module_batch(email_modules, entry.email, configs) + ) + round_results.extend(_tag_emails(email_results, entry)) + + checked.update((p.site, p.username.lower()) for p in pivots if p.site) + cross_results.extend(round_results) + + if round_number < depth: + _apply_confidence(results, cross_results, all_pivots, email_ratings) + source = _followable(round_results) + + _apply_confidence(results, cross_results, all_pivots, email_ratings) + _print_summary(results, cross_results) + return cross_results + + +def _scope( + cross_configs: CrossScanConfig, configs: ScanConfig +) -> Optional[List[ModuleType]]: + """The user_scan modules this run is allowed to touch, or None if unrestricted.""" + if cross_configs.modules: + return [ + module + for name in cross_configs.modules + for module in find_module( + name.replace(".", "_"), is_email=False, no_nsfw=configs.no_nsfw + ) + ] + if cross_configs.categories: + paths = load_categories(False, configs.no_nsfw) + return [ + module + for name in cross_configs.categories + if name in paths + for module in load_modules(paths[name]) + ] + return None + + +def _email_scope( + cross_configs: CrossScanConfig, configs: ScanConfig +) -> Optional[List[ModuleType]]: + """The email_scan modules this run may touch, or None for all of them. + + Loud modules are dropped rather than prompted for. The addresses reaching + here came off somebody else's profile, and mailing a third party is not a + decision a second scan pass should be making; ``--allow-loud`` puts them + back for a caller who has already accepted that. + """ + if cross_configs.emails == "none": + return [] + + modules: Optional[List[ModuleType]] = None + if cross_configs.modules: + modules = [ + module + for name in cross_configs.modules + for module in find_module( + name.replace(".", "_"), is_email=True, no_nsfw=configs.no_nsfw + ) + ] + elif cross_configs.categories: + paths = load_categories(True, configs.no_nsfw) + modules = [ + module + for name in cross_configs.categories + if name in paths + for module in load_modules(paths[name]) + ] + elif configs.allow_loud: + return None + else: + modules = [ + module + for path in load_categories(True, configs.no_nsfw).values() + for module in load_modules(path) + ] + + if not configs.allow_loud: + modules = [m for m in modules if not is_loud(get_site_name(m), is_email=True)] + return modules + + +def _already_swept(results: Iterable[Result]) -> Set[str]: + """Usernames the first pass already ran against every module. + + A username pass is itself a sweep of its own target, and sites tend to report + that handle straight back, so without this it would rank first and spend a + sweep repeating the scan that just finished. An email pass contributes + nothing here — its target is not a username. + """ + return {str(r.username).lower() for r in results if not r.is_email and r.username} + + +def _fresh_pivots( + source: List[Result], links: str, swept: Set[str], checked: Set[Tuple[str, str]] +) -> List[Pivot]: + """Pivots from ``source`` that no earlier round has already acted on. + + A swept username had every module run against it, so nothing about it is + left to do; a named pair is done once that one site has been checked. + """ + pivots = select_pivots(extract_pivots(source), links) + return [ + pivot + for pivot in pivots + if pivot.username.lower() not in swept + and (pivot.site is None or (pivot.site, pivot.username.lower()) not in checked) + ] + + +def _already_scanned(results: Iterable[Result]) -> Set[str]: + """Addresses the first pass already ran against every email module. + + An email pass is itself a scan of its own target, and profiles routinely + report that address straight back, so without this it would rank first and + spend the budget repeating the scan that just finished. + """ + return {str(r.username).lower() for r in results if r.is_email and r.username} + + +def _fresh_emails(source: List[Result], emails: str, scanned: Set[str]) -> List[RankedEmail]: + """Addresses in ``source`` no earlier round has already scanned, best first.""" + pivots = [ + pivot + for pivot in select_email_pivots(extract_email_pivots(source), emails) + if pivot.email not in scanned + ] + if not pivots: + return [] + return rank_emails(pivots, build_anchors(confirmed=_followable(source))) + + +def _followable(round_results: List[Result]) -> List[Result]: + """The hits a further round may pivot off. + + An account whose metadata names someone else is a handle collision, and + following its links would walk into a stranger's footprint. + """ + return [ + result + for result in round_results + if result.is_found() + and result.extra.get("confidence") != Confidence.CONFLICTING.value + ] + + +def _split_budget(budget: int, usernames: int, emails: int) -> Tuple[int, int]: + """Share one sweep budget between the two target kinds. + + Half is offered to addresses, rounded down so a budget of 1 still sweeps a + username — the behaviour before addresses existed — and whatever one kind + cannot use falls to the other. Neither can starve the other outright: with + any budget at all, both get a slot as soon as there are two to give. + """ + if budget <= 0: + return 0, 0 + for_emails = min(budget // 2, emails) + for_usernames = min(budget - for_emails, usernames) + return for_usernames, min(budget - for_usernames, emails) + + +def _sweep_targets(pivots: List[Pivot], budget: int) -> List[str]: + ranked = rank_usernames(pivots) + + if budget <= 0: + # Only a username no pivot ever tied to a site goes unchecked; one that + # names a site elsewhere is still reached by that named check. + routed = {p.username.lower() for p in pivots if p.site} + unreachable = sorted({p.username for p in pivots if p.username.lower() not in routed}) + print( + f"{Fore.YELLOW}[!] Username sweep off — checking only the sites pivots " + f"named.{Style.RESET_ALL}" + ) + if unreachable: + print( + f"{Fore.YELLOW}[!] {len(unreachable)} pivot username(s) name no site " + f"({', '.join(unreachable)}) and are not checked{Style.RESET_ALL}" + ) + return [] + + usernames = ranked[:budget] + deferred = ranked[len(usernames) :] + if deferred: + print( + f"{Fore.YELLOW}[!] Not sweeping {len(deferred)} username(s) " + f"({', '.join(deferred)}) — raise --cross-sweep{Style.RESET_ALL}" + ) + return usernames + + +def _email_targets( + ranked: List[RankedEmail], budget: int, modules: Optional[List[ModuleType]] +) -> List[RankedEmail]: + if modules is not None and not modules: + if ranked: + print( + f"{Fore.YELLOW}[!] The -m/-c restriction names no email module, so " + f"no address is scanned.{Style.RESET_ALL}" + ) + return [] + + if budget <= 0: + if ranked: + print( + f"{Fore.YELLOW}[!] No sweep budget left for {len(ranked)} address(es) " + f"— raise --cross-sweep{Style.RESET_ALL}" + ) + return [] + + taken = ranked[:budget] + deferred = ranked[len(taken) :] + if deferred: + print( + f"{Fore.YELLOW}[!] Not scanning {len(deferred)} address(es) " + f"({', '.join(entry.email for entry in deferred)}) — raise " + f"--cross-sweep{Style.RESET_ALL}" + ) + return taken + + +def _print_emails(ranked: List[RankedEmail]) -> None: + print(f"{Fore.GREEN}[+] {len(ranked)} address(es) extracted{Style.RESET_ALL}") + for entry in ranked: + color = _CONFIDENCE_COLOR[entry.confidence] + print( + f" {color}{entry.confidence.value:<12}{Style.RESET_ALL}" + f"{entry.email:<32} {Fore.WHITE}{', '.join(entry.sources)}{Style.RESET_ALL}" + ) + + +def _print_pivots(pivots: List[Pivot]) -> None: + print(f"{Fore.GREEN}[+] {len(pivots)} pivot(s) extracted{Style.RESET_ALL}") + for pivot in pivots: + tag = f"[{pivot.kind.value}]" + print( + f" {Fore.CYAN}{tag:<11}{Style.RESET_ALL}" + f"{pivot.username:<24} {(pivot.site or '(any site)'):<16} " + f"{Fore.WHITE}{pivot.origin}{Style.RESET_ALL}" + ) + + +def _named_targets( + pivots: Iterable[Pivot], + swept: Set[str], + checked: Set[Tuple[str, str]], + configs: ScanConfig, + scope: Optional[List[ModuleType]] = None, +) -> Dict[str, List[ModuleType]]: + """Modules to check for pivots whose username was not swept. + + A swept username already ran against every module, so only the pivots left + over need their one named site checked. + """ + # Grouped case-insensitively, matching how `swept` and `checked` compare: + # two profiles can link the same account in different cases, and scanning it + # once per casing is a duplicate request for one account. Pivots arrive + # best-vouched first, so the first casing seen is the one worth reporting. + sites_by_username: Dict[str, Set[str]] = {} + casing: Dict[str, str] = {} + for pivot in pivots: + key = pivot.username.lower() + if not pivot.site or key in swept or (pivot.site, key) in checked: + continue + casing.setdefault(key, pivot.username) + sites_by_username.setdefault(key, set()).add(pivot.site) + + allowed = None if scope is None else {module.__name__ for module in scope} + + targets: Dict[str, List[ModuleType]] = {} + for key, sites in sites_by_username.items(): + if allowed is not None: + sites = {site for site in sites if site in allowed} + username = casing[key] + modules = [ + module + for site in sorted(sites) + for module in find_module(site, is_email=False, no_nsfw=configs.no_nsfw) + ] + if modules: + targets[username] = modules + return targets + + +def _tag(results: List[Result], pivots: Iterable[Pivot], username: str) -> List[Result]: + """Record on each hit why its username was scanned. + + A username often arrives from several pivots at once; only the best-vouched + class is reported, since that is the claim the scan rests on. + """ + matching = [pivot for pivot in pivots if pivot.username.lower() == username.lower()] + if not matching: + return results + + best = min(pivot.kind.rank for pivot in matching) + strongest = [pivot for pivot in matching if pivot.kind.rank == best] + label = f"{strongest[0].kind.value} from {', '.join(sorted({p.origin for p in strongest}))}" + + for result in results: + if result.is_found(): + result.update(extra={"pivot_source": label}) + return results + + +def _tag_emails(results: List[Result], entry: RankedEmail) -> List[Result]: + """Record on each hit which profile published the address that found it.""" + label = f"address from {', '.join(entry.sources)}" + for result in results: + if result.is_found(): + result.update(extra={"pivot_source": label}) + return results + + +def _apply_confidence( + prior: Iterable[Result], + cross_results: Iterable[Result], + pivots: Iterable[Pivot], + email_ratings: Dict[str, Confidence], +) -> None: + """Rate every hit, and record the rating on it. + + Scoring runs once the pass is over because the anchors come from its own + confirmed hits — a sweep hit cannot be judged before the accounts it is + judged against have been fetched. + + An account reached by scanning an address inherits that address's rating + rather than being scored on its own metadata: the account is only as well + tied to the target as the address that led to it, and an email module's + verdict says nothing about who owns the mailbox. + """ + named = {(p.site, p.username.lower()) for p in pivots if p.site} + hits = [r for r in cross_results if r.is_found()] + anchors = build_anchors( + confirmed=[r for r in hits if _is_named(r, named)], + emails=[str(r.username) for r in prior if r.is_email and r.username], + urls=[p.url for p in pivots if p.url], + ) + + for result in hits: + inherited = email_ratings.get(str(result.username or "").lower()) if result.is_email else None + rating = inherited or score(result, anchors, confirmed=_is_named(result, named)) + result.update(extra={"confidence": rating.value}) + + +def _is_named(result: Result, named: Set[Tuple[str, str]]) -> bool: + stem = module_stem(str(result.site_name or "")) + return bool(stem) and (stem, str(result.username or "").lower()) in named + + +def _print_summary(prior: Iterable[Result], cross_results: Iterable[Result]) -> None: + hits = [r for r in cross_results if r.is_found()] + print(f"\n{Fore.CYAN}[i] Cross-scan complete.{Style.RESET_ALL}") + print(f" Accounts found: {len(hits)}") + + addresses = sorted({str(r.username) for r in hits if r.is_email and r.username}) + if addresses: + print(f" Reached via address: {', '.join(addresses)}") + + by_rating = {rating: _sites_rated(hits, rating) for rating in ORDER} + for rating in ORDER: + sites = by_rating[rating] + if not sites: + continue + color = _CONFIDENCE_COLOR[rating] + print(f" {color}{rating.value:<12}{Style.RESET_ALL}{len(sites):>4} {rating.explanation}") + + for rating in (Confidence.CONFIRMED, Confidence.LIKELY, Confidence.CONFLICTING): + sites = by_rating[rating] + if sites: + color = _CONFIDENCE_COLOR[rating] + print(f"\n {color}{rating.value}:{Style.RESET_ALL} {', '.join(sites)}") + + prior_sites = {str(r.site_name).lower() for r in prior if r.is_found() and r.site_name} + new_sites = sorted( + { + str(r.site_name) + for r in hits + if r.site_name and str(r.site_name).lower() not in prior_sites + } + ) + if not new_sites: + print(f"\n {Fore.YELLOW}No sites beyond the first pass.{Style.RESET_ALL}") + return + print(f"\n {Fore.GREEN}New sites the first pass missed ({len(new_sites)}):{Style.RESET_ALL}") + print(f" {', '.join(new_sites)}") + + +def _sites_rated(hits: Iterable[Result], rating: Confidence) -> List[str]: + return sorted( + f"{r.site_name} ({r.username})" + for r in hits + if r.extra.get("confidence") == rating.value + ) diff --git a/user_scanner/core/pdf_generator.py b/user_scanner/core/pdf_generator.py index 8c6a69a9..7ae4edf0 100644 --- a/user_scanner/core/pdf_generator.py +++ b/user_scanner/core/pdf_generator.py @@ -416,67 +416,133 @@ def generate_pdf_report( # Footprint Table elements.append(Paragraph("DIGITAL FOOTPRINT MAPPING", section_title_style)) - table_data = [ - [ - Paragraph( - "SL", - normal_style, - ), - Paragraph( - "PLATFORM", - normal_style, - ), - Paragraph( - "CATEGORY", - normal_style, - ), - Paragraph( - "STATUS", - normal_style, - ), - Paragraph( - "IDENTIFIED URL", - normal_style, - ), - ] - ] + is_cross_scan = "cross" in scan_type.lower() - for idx, hit in enumerate(hits): - status = str(hit.get("status", "Unknown")) - status_color = ( - "#16a34a" - if status.lower() in ["found", "registered", "taken"] - else "#111111" - ) - url_text = truncate(hit.get("url", "N/A"), 100) - - table_data.append( + if is_cross_scan: + table_data = [ [ Paragraph( - f"{idx + 1}", - ParagraphStyle("C", alignment=1), + "SL", + normal_style, + ), + Paragraph( + "TARGET", + normal_style, ), Paragraph( - f"{html.escape(str(hit.get('site_name', '')))}", normal_style + "PLATFORM", + normal_style, ), Paragraph( - f"{html.escape(str(hit.get('category', '')))}", normal_style + "CATEGORY", + normal_style, ), Paragraph( - f"{html.escape(status)}", + "STATUS", normal_style, ), Paragraph( - f"{html.escape(url_text)}", normal_style + "IDENTIFIED URL", + normal_style, ), ] + ] + else: + table_data = [ + [ + Paragraph( + "SL", + normal_style, + ), + Paragraph( + "PLATFORM", + normal_style, + ), + Paragraph( + "CATEGORY", + normal_style, + ), + Paragraph( + "STATUS", + normal_style, + ), + Paragraph( + "IDENTIFIED URL", + normal_style, + ), + ] + ] + + for idx, hit in enumerate(hits): + status = str(hit.get("status", "Unknown")) + status_color = ( + "#16a34a" + if status.lower() in ["found", "registered", "taken"] + else "#111111" ) + url_text = truncate(hit.get("url", "N/A"), 100) + target_val = html.escape(str(hit.get("email") or hit.get("username") or "")) + + if is_cross_scan: + table_data.append( + [ + Paragraph( + f"{idx + 1}", + ParagraphStyle("C", alignment=1), + ), + Paragraph( + f"{target_val}", normal_style + ), + Paragraph( + f"{html.escape(str(hit.get('site_name', '')))}", normal_style + ), + Paragraph( + f"{html.escape(str(hit.get('category', '')))}", normal_style + ), + Paragraph( + f"{html.escape(status)}", + normal_style, + ), + Paragraph( + f"{html.escape(url_text)}", normal_style + ), + ] + ) + else: + table_data.append( + [ + Paragraph( + f"{idx + 1}", + ParagraphStyle("C", alignment=1), + ), + Paragraph( + f"{html.escape(str(hit.get('site_name', '')))}", normal_style + ), + Paragraph( + f"{html.escape(str(hit.get('category', '')))}", normal_style + ), + Paragraph( + f"{html.escape(status)}", + normal_style, + ), + Paragraph( + f"{html.escape(url_text)}", normal_style + ), + ] + ) - data_table = Table( - table_data, - colWidths=[0.5 * inch, 1.5 * inch, 1.2 * inch, 1.0 * inch, 3.1 * inch], - repeatRows=1, - ) + if is_cross_scan: + data_table = Table( + table_data, + colWidths=[0.4 * inch, 1.5 * inch, 1.2 * inch, 0.9 * inch, 0.9 * inch, 2.4 * inch], + repeatRows=1, + ) + else: + data_table = Table( + table_data, + colWidths=[0.5 * inch, 1.5 * inch, 1.2 * inch, 1.0 * inch, 3.1 * inch], + repeatRows=1, + ) data_table.setStyle( TableStyle( [ @@ -509,9 +575,14 @@ def generate_pdf_report( ) for hit, meta in hits_with_meta: + site_title = html.escape(str(hit.get('site_name', ''))) + target_val = html.escape(str(hit.get("email") or hit.get("username") or "")) + if is_cross_scan and target_val: + site_title += f" ({target_val})" + meta_elements = [ Paragraph( - f"{html.escape(str(hit.get('site_name', '')))}", + f"{site_title}", normal_style, ), Spacer(1, 5), diff --git a/user_scanner/core/pivots.py b/user_scanner/core/pivots.py new file mode 100644 index 00000000..b8585c86 --- /dev/null +++ b/user_scanner/core/pivots.py @@ -0,0 +1,644 @@ +"""Turns finished scan metadata into new scan targets. + +A pivot is a target a scan implied but never tested: a handle a site reports +for the scanned email, one embedded in a link that profile carries, or an email +address the profile publishes. Nothing here makes a request — callers decide +which pivots are worth one. + +Links are classified by how much the source platform vouches for them. +A *verified* link required the owner to prove control of the far side (OAuth +connection, rel="me" round-trip); a plain *link* is free text the owner typed. +Email addresses carry the same distinction: one the site published in its own +email field against one scraped out of prose the owner wrote. +""" + +import re +from dataclasses import dataclass +from enum import Enum +from typing import Iterable, Iterator, List, Optional, Tuple +from urllib.parse import unquote, urlsplit + +from user_scanner.core.helpers import EMAIL_RE, is_valid_email +from user_scanner.core.result import Result + +# Extra keys whose value is the account's own handle on the reporting site. +HANDLE_KEYS = frozenset( + { + "username", + "user_name", + "handle", + "screen_name", + "nickname", + "login", + "login_name", + "preferred_username", + "profile_name", + "vanity", + } +) + +# Extra keys whose links the platform itself verified. +VERIFIED_KEYS = frozenset({"verified_accounts", "verified_links", "connected_accounts"}) + +# Extra keys named after another platform, holding a bare handle rather than a +# URL. Kick, GitHub, Unsplash, Coderwall and 500px all publish socials this way, +# and a bare handle never reaches ``_pivots_from_links``, which only matches URLs. +# +# ``discord`` is deliberately absent: sites store a server invite code there +# (Kick's ``nAZEkUNWPt``), which is not an account name. +PLATFORM_HANDLE_KEYS = { + "bluesky": "bluesky", + "facebook": "facebook", + "github": "github", + "instagram": "instagram", + "linkedin": "linkedin", + "mastodon": "mastodon", + "pinterest": "pinterest", + "reddit": "reddit", + "soundcloud": "soundcloud", + "spotify": "spotify", + "tiktok": "tiktok", + "tumblr": "tumblr", + "twitch": "twitch", + "twitter": "x", + "x": "x", + "youtube": "youtube", +} + +# Suffixes a site may hang off a platform name (``twitter_handle``). ``_id`` is +# excluded on purpose so that identifier fields — YouTube's ``UC…`` channel id — +# are never read as handles. +_PLATFORM_KEY_SUFFIXES = ("_handle", "_username", "_user", "_name") + +# Extra keys whose value is the account holder's own address. A key outside this +# set may still carry an address, but only as free text — which is what keeps +# package metadata (``author_email``, ``maintainer_email``) out of the trusted +# tier, since those name whoever published a release rather than the account. +EMAIL_KEYS = frozenset( + { + "email", + "emails", + "business_email", + "contact_email", + "public_email", + "paypal_email", + "verified_email", + } +) + +# Local parts that address a role or a robot rather than a person. Kept to the +# RFC 2142 mandated names and the no-reply family: a freelancer really does take +# mail at ``hello@`` or ``contact@``, so those stay in. +_ROLE_LOCAL_PARTS = frozenset( + { + "abuse", + "do-not-reply", + "donotreply", + "hostmaster", + "mailer-daemon", + "no-reply", + "noreply", + "postmaster", + "webmaster", + } +) + +# Domains that hold no mailbox: RFC 2606 documentation placeholders, the +# stand-ins people type into a bio, and the relay GitHub substitutes when an +# account hides its address. +_NON_MAILBOX_DOMAINS = frozenset( + { + "domain.com", + "email.com", + "example.com", + "example.net", + "example.org", + "users.noreply.github.com", + "yourdomain.com", + } +) + +_NON_MAILBOX_TLDS = (".example", ".invalid", ".localhost", ".test") + +# Substrings marking a value as an image URL — never a pivot. +_MEDIA_KEY_PARTS = ( + "avatar", + "image", + "photo", + "picture", + "thumbnail", + "icon", + "banner", + "background", + "snapcode", + "pfp", + "logo", +) + +_URL_RE = re.compile(r"https?://[^\s,;\"'<>()\[\]]+", re.I) +_VERIFIED_SUFFIX_RE = re.compile(r"\s*\(verified\)", re.I) +_HANDLE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._~-]{1,63}$") + +# Path segments and domain labels that name a site's own pages rather than a +# person, so a bare-segment route must never hand them back as a handle. +_RESERVED = frozenset( + { + "about", "account", "accounts", "admin", "all", "api", "assets", "auth", + "best", "blog", "categories", "category", "channel", "channels", "contact", + "cookies", "dashboard", "developer", "developers", "discover", "docs", + "download", "downloads", "embed", "en", "event", "events", "explore", + "faq", "feed", "feeds", "features", "forum", "forums", "group", "groups", + "help", "home", "i", "images", "img", "inbox", "index", "intent", "jobs", + "join", "learn", "legal", "login", "logout", "mail", "media", "messages", + "new", "news", "notifications", "oauth", "p", "page", "pages", "playlist", + "popular", "press", "pricing", "privacy", "products", "profile", "public", + "register", "results", "rss", "search", "secure", + "settings", "share", "shop", "signin", "signup", "sitemap", "static", + "store", "support", "tag", "tags", "terms", "top", "topic", "topics", + "tos", "trending", "us", "user", "users", "video", "videos", "watch", + "web", "www", + } +) + +_BARE = r"^/(?P[^/?#]+)/?$" +_AT = r"^/@(?P[^/?#]+)/?$" + +# host(s) -> user_scan module, path patterns yielding the handle. Every module +# named here must exist under user_scan/. +_HOST_ROUTES: Tuple[Tuple[Tuple[str, ...], str, Tuple[str, ...]], ...] = ( + (("x.com", "twitter.com"), "x", (_BARE,)), + (("linkedin.com",), "linkedin", (r"^/in/(?P[^/?#]+)/?$",)), + (("github.com",), "github", (_BARE,)), + (("gist.github.com",), "githubgist", (_BARE,)), + (("gitlab.com",), "gitlab", (_BARE,)), + (("bitbucket.org",), "bitbucket", (_BARE,)), + (("codeberg.org",), "codeberg", (_BARE,)), + (("gitee.com",), "gitee", (_BARE,)), + (("stackoverflow.com",), "stackoverflow", (r"^/users/\d+/(?P[^/?#]+)/?$",)), + ( + ("youtube.com", "youtu.be"), + "youtube", + (_AT, r"^/c/(?P[^/?#]+)/?$", r"^/user/(?P[^/?#]+)/?$"), + ), + (("instagram.com",), "instagram", (_BARE,)), + (("facebook.com", "fb.com"), "facebook", (_BARE,)), + (("threads.net", "threads.com"), "threads", (_AT,)), + (("tiktok.com",), "tiktok", (_AT,)), + (("reddit.com",), "reddit", (r"^/u(?:ser)?/(?P[^/?#]+)/?$",)), + (("mastodon.social",), "mastodon", (_AT,)), + (("bsky.app",), "bluesky", (r"^/profile/(?P[^/?#]+)/?$",)), + (("t.me", "telegram.me"), "telegram", (_BARE,)), + (("twitch.tv",), "twitch", (_BARE,)), + (("keybase.io",), "keybase", (_BARE,)), + (("vk.com",), "vk", (_BARE,)), + (("pinterest.com",), "pinterest", (_BARE,)), + (("tumblr.com",), "tumblr", (_BARE,)), + (("medium.com",), "medium", (_AT,)), + (("dev.to",), "devto", (_BARE,)), + (("hashnode.com",), "hashnode", (_AT,)), + (("substack.com",), "substack", (_AT,)), + (("about.me",), "about_me", (_BARE,)), + (("linktr.ee",), "linktree", (_BARE,)), + (("gravatar.com",), "gravatar", (_BARE,)), + (("behance.net",), "behance", (_BARE,)), + (("dribbble.com",), "dribbble", (_BARE,)), + (("deviantart.com",), "deviantart", (_BARE,)), + (("unsplash.com",), "unsplash", (_AT,)), + (("flickr.com",), "flickr", (r"^/(?:photos|people)/(?P[^/?#]+)/?$",)), + (("vimeo.com",), "vimeo", (_BARE,)), + (("dailymotion.com",), "dailymotion", (_BARE,)), + (("soundcloud.com",), "soundcloud", (_BARE,)), + (("mixcloud.com",), "mixcloud", (_BARE,)), + (("last.fm",), "lastfm", (r"^/user/(?P[^/?#]+)/?$",)), + # Listed with no path pattern on purpose: /user/ carries a base62 id Spotify + # assigns, which is portable nowhere and is not what the spotify module looks + # up anyway (it queries stats.fm handles). Keeping the host routed still marks + # it as a platform rather than someone's personal domain. + (("open.spotify.com",), "spotify", ()), + (("bandlab.com",), "bandlab", (_BARE,)), + (("discogs.com",), "discogs", (r"^/user/(?P[^/?#]+)/?$",)), + (("npmjs.com",), "npmjs", (r"^/~(?P[^/?#]+)/?$",)), + (("pypi.org",), "pypi", (r"^/user/(?P[^/?#]+)/?$",)), + (("crates.io",), "cratesio", (r"^/users/(?P[^/?#]+)/?$",)), + (("rubygems.org",), "rubygems", (r"^/profiles/(?P[^/?#]+)/?$",)), + (("hub.docker.com",), "dockerhub", (r"^/u/(?P[^/?#]+)/?$",)), + (("huggingface.co",), "huggingface", (_BARE,)), + (("kaggle.com",), "kaggle", (_BARE,)), + (("hackerrank.com",), "hackerrank", (r"^/(?:profile/)?(?P[^/?#]+)/?$",)), + (("hackerone.com",), "hackerone", (_BARE,)), + (("leetcode.com",), "leetcode", (r"^/u/(?P[^/?#]+)/?$",)), + (("codeforces.com",), "codeforces", (r"^/profile/(?P[^/?#]+)/?$",)), + (("codewars.com",), "codewars", (r"^/users/(?P[^/?#]+)/?$",)), + (("scratch.mit.edu",), "scratch", (r"^/users/(?P[^/?#]+)/?$",)), + (("figma.com",), "figma", (_AT,)), + (("producthunt.com",), "producthunt", (_AT,)), + (("patreon.com",), "patreon", (_BARE,)), + (("ko-fi.com", "buymeacoffee.com"), "buymeacoffee", (_BARE,)), + (("liberapay.com",), "liberapay", (_BARE,)), + (("goodreads.com",), "goodreads", (r"^/user/show/\d+-(?P[^/?#]+)/?$",)), + (("myanimelist.net",), "myanimelist", (r"^/profile/(?P[^/?#]+)/?$",)), + (("anilist.co",), "anilist", (r"^/user/(?P[^/?#]+)/?$",)), + (("chess.com",), "chess_com", (r"^/member/(?P[^/?#]+)/?$",)), + (("lichess.org",), "lichess", (r"^/@/(?P[^/?#]+)/?$",)), + (("steamcommunity.com",), "steam", (r"^/id/(?P[^/?#]+)/?$",)), + (("speedrun.com",), "speedrun", (r"^/(?:user/)?(?P[^/?#]+)/?$",)), + (("osu.ppy.sh",), "osu", (r"^/users/(?P[^/?#]+)/?$",)), + (("trello.com",), "trello", (r"^/u/(?P[^/?#]+)/?$",)), + (("imgur.com",), "imgur", (r"^/user/(?P[^/?#]+)/?$",)), + (("giphy.com",), "giphy", (r"^/channel/(?P[^/?#]+)/?$",)), + (("speakerdeck.com",), "speakerdeck", (_BARE,)), + (("issuu.com",), "issuu", (_BARE,)), + (("fiverr.com",), "fiverr", (_BARE,)), + (("calendly.com",), "calendly", (_BARE,)), + (("tradingview.com",), "tradingview", (r"^/u/(?P[^/?#]+)/?$",)), + (("openstreetmap.org",), "openstreetmap", (r"^/user/(?P[^/?#]+)/?$",)), + (("tripadvisor.com",), "tripadvisor", (r"^/Profile/(?P[^/?#]+)/?$",)), + (("duolingo.com",), "duolingo", (r"^/profile/(?P[^/?#]+)/?$",)), +) + +# .host -> user_scan module. +_SUBDOMAIN_ROUTES: Tuple[Tuple[str, str], ...] = ( + ("tumblr.com", "tumblr"), + ("wordpress.com", "wordpress"), + ("blogspot.com", "blogger"), + ("medium.com", "medium"), + ("substack.com", "substack"), + ("bandcamp.com", "bandcamp"), + ("itch.io", "itch_io"), + ("carrd.co", "carrd"), + ("github.io", "github"), + ("hashnode.dev", "hashnode"), + ("gitbook.io", "gitbook"), + ("weebly.com", "weebly"), + ("wixsite.com", "wix"), + ("deviantart.com", "deviantart"), +) + +_ROUTES = { + host: (module, tuple(re.compile(p) for p in patterns)) + for hosts, module, patterns in _HOST_ROUTES + for host in hosts +} + + +class PivotKind(Enum): + """How strongly the reporting platform vouches for a pivot.""" + + HANDLE = "handle" + VERIFIED = "verified" + LINK = "link" + + @property + def rank(self) -> int: + return _KIND_RANK[self] + + +_KIND_RANK = {PivotKind.HANDLE: 0, PivotKind.VERIFIED: 1, PivotKind.LINK: 2} + + +@dataclass(frozen=True) +class Pivot: + """A username worth scanning, and where it came from.""" + + username: str + kind: PivotKind + source_site: str + source_key: str + site: Optional[str] = None + url: str = "" + + @property + def origin(self) -> str: + return f"{self.source_site} ({self.source_key})" + + +class EmailKind(Enum): + """How the source platform presented an address. + + ``FIELD`` is the site's own email field for the account; ``TEXT`` is an + address read out of prose, where nothing says whose mailbox it is. + """ + + FIELD = "field" + TEXT = "text" + + @property + def rank(self) -> int: + return 0 if self is EmailKind.FIELD else 1 + + +@dataclass(frozen=True) +class EmailPivot: + """An address worth scanning, and where it came from.""" + + email: str + kind: EmailKind + source_site: str + source_key: str + + @property + def origin(self) -> str: + return f"{self.source_site} ({self.source_key})" + + +def extract_pivots(results: Iterable[Result]) -> List[Pivot]: + """Collect every username a set of finished results implies. + + Only results that found an account are mined — a miss carries no metadata + worth following. + """ + pivots: List[Pivot] = [] + seen = set() + + for result in results: + if not result.is_found(): + continue + for pivot in _pivots_from_result(result): + key = (pivot.username.lower(), pivot.site, pivot.kind) + if key in seen: + continue + seen.add(key) + pivots.append(pivot) + + return sorted(pivots, key=lambda p: (p.kind.rank, p.source_site, p.username.lower())) + + +def select_pivots(pivots: Iterable[Pivot], links: str = "all") -> List[Pivot]: + """Filter pivots by link class. + + ``all`` keeps everything, ``verified`` drops owner-entered links, ``none`` + keeps only handles the source site reported itself. + """ + if links == "verified": + return [p for p in pivots if p.kind is not PivotKind.LINK] + if links == "none": + return [p for p in pivots if p.kind is PivotKind.HANDLE] + return list(pivots) + + +def extract_email_pivots(results: Iterable[Result]) -> List[EmailPivot]: + """Collect every address a set of finished results exposes. + + One pivot per address *per source site*, so a caller can tell an address two + profiles agree on from one only a single profile mentions. + """ + pivots: List[EmailPivot] = [] + seen = set() + + for result in results: + if not result.is_found(): + continue + for pivot in _email_pivots_from_result(result): + key = (pivot.email, pivot.source_site, pivot.kind) + if key in seen: + continue + seen.add(key) + pivots.append(pivot) + + return sorted(pivots, key=lambda p: (p.kind.rank, p.source_site, p.email)) + + +def select_email_pivots(pivots: Iterable[EmailPivot], emails: str = "verified") -> List[EmailPivot]: + """Filter address pivots by how the source presented them. + + ``all`` keeps addresses scraped from prose too, ``verified`` keeps only the + ones a site published in its own email field, and ``none`` keeps nothing — + unlike its ``--cross-links`` namesake, which still yields handle pivots + because a handle is not a link. Every tier here is an address. + """ + if emails == "none": + return [] + if emails == "all": + return list(pivots) + return [pivot for pivot in pivots if pivot.kind is EmailKind.FIELD] + + +def rank_usernames(pivots: Iterable[Pivot]) -> List[str]: + """Order distinct usernames by how well vouched they are, then by how many + pivots mention them.""" + best: dict = {} + + for pivot in pivots: + key = pivot.username.lower() + entry = best.get(key) + if entry is None: + best[key] = {"username": pivot.username, "rank": pivot.kind.rank, "count": 1} + continue + entry["count"] += 1 + if pivot.kind.rank < entry["rank"]: + entry["rank"] = pivot.kind.rank + entry["username"] = pivot.username + + return [ + entry["username"] + for entry in sorted( + best.values(), key=lambda e: (e["rank"], -e["count"], e["username"].lower()) + ) + ] + + +def resolve_url(url: str) -> Tuple[Optional[str], Optional[str]]: + """Map a profile URL to ``(user_scan module, username)``. + + Returns ``(None, username)`` for a personal domain, whose handle is worth + scanning everywhere but points at no particular site, and ``(None, None)`` + when nothing usable can be read out. + """ + try: + parts = urlsplit(url) + except ValueError: + return None, None + + if parts.scheme not in ("http", "https") or not parts.netloc: + return None, None + + host = parts.hostname or "" + path = unquote(parts.path or "") + + subdomain = _match_subdomain(host) + if subdomain: + return subdomain + + for candidate in _host_candidates(host): + route = _ROUTES.get(candidate) + if not route: + continue + module, patterns = route + for pattern in patterns: + match = pattern.match(path) + if match: + user = _clean_handle(match.group("user"), module) + if user: + return module, user + # A routed host whose path fits no profile shape is one of that site's + # own pages, so the domain fallback must not fire for it. + return None, None + + return None, _domain_handle(host, path) + + +def is_platform_host(host: str) -> bool: + """True when a host belongs to a site the route table knows. + + Lets callers tell a personal domain from a platform without caring whether + any particular URL on it happens to name a profile. + """ + host = (host or "").strip().lower().rstrip(".").removeprefix("www.") + if not host: + return False + if any(host == suffix or host.endswith("." + suffix) for suffix, _ in _SUBDOMAIN_ROUTES): + return True + return any(candidate in _ROUTES for candidate in _host_candidates(host)) + + +def _pivots_from_result(result: Result) -> Iterator[Pivot]: + source_site = result.site_name or "Unknown" + own_module = module_stem(source_site) + + for key, value in result.extra.items(): + if not isinstance(value, str) or is_media_key(key): + continue + + if key in HANDLE_KEYS: + handle = _clean_handle(value, own_module) + if handle: + yield Pivot(handle, PivotKind.HANDLE, source_site, key, own_module) + continue + + module = _platform_key_module(key) + if module and not _URL_RE.search(value): + handle = _clean_handle(value, module) + if handle: + yield Pivot(handle, PivotKind.LINK, source_site, key, module) + continue + + yield from _pivots_from_links(value, key, source_site, own_module) + + +def _platform_key_module(key: str) -> Optional[str]: + """The module a platform-named key points at — ``twitter_handle`` -> ``x``. + + Only the key names the platform; the value is the handle. Returns ``None`` + for any key that does not name a platform, so an unknown key still falls + through to URL extraction. + """ + name = key.lower() + for suffix in _PLATFORM_KEY_SUFFIXES: + name = name.removesuffix(suffix) + return PLATFORM_HANDLE_KEYS.get(name) + + +def _email_pivots_from_result(result: Result) -> Iterator[EmailPivot]: + source_site = result.site_name or "Unknown" + + for key, value in result.extra.items(): + if not isinstance(value, str) or is_media_key(key): + continue + kind = EmailKind.FIELD if key in EMAIL_KEYS else EmailKind.TEXT + for candidate in _addresses_in(value): + address = _clean_email(candidate) + if address: + yield EmailPivot(address, kind, source_site, key) + + +def _addresses_in(text: str) -> Iterator[str]: + """Addresses in a value, minus the two shapes that only look like one. + + URLs are blanked first, because a profile link parses as a perfectly legal + address whose local part is the host and path — ``tiktok.com/@jane.doe`` + yields ``//www.tiktok.com/@jane.doe``. Nothing is lost: links already + reach the scan as username pivots. + + A match the text prefixes with a second ``@`` is a fediverse handle + (``@johndoe@mastodon.social``), which is an account name, not a mailbox. + """ + masked = _URL_RE.sub(lambda m: " " * len(m.group(0)), text) + for match in EMAIL_RE.finditer(masked): + if match.start() and masked[match.start() - 1] == "@": + continue + yield match.group(0) + + +def _clean_email(value: str) -> Optional[str]: + address = value.strip().strip(".,;:").lower() + if not is_valid_email(address): + return None + + local, _, domain = address.rpartition("@") + if local in _ROLE_LOCAL_PARTS or domain in _NON_MAILBOX_DOMAINS: + return None + if domain.startswith("noreply.") or ".noreply." in domain: + return None + if domain.endswith(_NON_MAILBOX_TLDS): + return None + return address + + +def _pivots_from_links( + value: str, key: str, source_site: str, own_module: Optional[str] +) -> Iterator[Pivot]: + for match in _URL_RE.finditer(value): + url = match.group(0).rstrip(".,;:") + verified = key in VERIFIED_KEYS or bool(_VERIFIED_SUFFIX_RE.match(value, match.end())) + module, username = resolve_url(url) + if not username: + continue + # A site's own address inside its own metadata names the site, not a person. + if module is None and own_module and username.lower() == own_module: + continue + kind = PivotKind.VERIFIED if verified else PivotKind.LINK + yield Pivot(username, kind, source_site, key, module, url) + + +def _match_subdomain(host: str) -> Optional[Tuple[str, str]]: + for suffix, module in _SUBDOMAIN_ROUTES: + if not host.endswith("." + suffix): + continue + label = host[: -len(suffix) - 1] + if "." in label: + continue + user = _clean_handle(label, module) + if user: + return module, user + return None + + +def _host_candidates(host: str) -> Iterator[str]: + """Yield the host and each parent domain, so country and www prefixes + (``br.linkedin.com``) reach the same route as the bare domain.""" + labels = host.split(".") + for index in range(len(labels) - 1): + yield ".".join(labels[index:]) + + +def _domain_handle(host: str, path: str) -> Optional[str]: + """Read a handle off a personal domain — only from its root page, where the + domain label is the whole claim being made.""" + if path not in ("", "/"): + return None + labels = host.split(".") + if len(labels) < 2: + return None + return _clean_handle(labels[-2], None) + + +def _clean_handle(value: str, module: Optional[str]) -> Optional[str]: + handle = unquote(value or "").strip().strip("@").rstrip("/") + if module == "bluesky": + handle = handle.removesuffix(".bsky.social") + if not _HANDLE_RE.match(handle) or handle.isdigit(): + return None + if handle.lower() in _RESERVED: + return None + return handle + + +def module_stem(site_name: str) -> Optional[str]: + """Reverse of ``helpers.get_site_name`` — the module file a site came from.""" + name = re.sub(r"\s*\(.*\)\s*$", "", (site_name or "").strip().lower()) + stem = re.sub(r"[^a-z0-9]+", "_", name).strip("_") + return stem or None + + +def is_media_key(key: str) -> bool: + return any(part in key for part in _MEDIA_KEY_PARTS)