2026-03-01 05:56:57
File: .gitignore
config.toml
# Scan artifacts
screenshots/
+# GeoIP data files
+*.csv.gz
+
# OS
.DS_Store
Thumbs.db
File: backend/atproto_client.py
def announce_host(self, host: dict, screenshot_path: Optional[str] = None, proto
hostname = host.get("hostname", "").strip()
hostname_suffix = f": {hostname}" if hostname else ""
asn = host.get("asn", "").strip()
+ isp = host.get("isp", "").strip()
ip_type = host.get("ip_type", "").strip()
+ city = host.get("city", "").strip()
+ country_code = host.get("country_code", "").strip()
+
+ # Build full location string: "Hayward / US / AS6167 Verizon Business"
+ location_parts = []
+ if city:
+ location_parts.append(city)
+ if country_code:
+ location_parts.append(country_code)
+ asn_full = f"{asn} {isp}".strip() if asn else isp
+ if asn_full:
+ location_parts.append(asn_full)
+ location = " / ".join(location_parts)
text = self.config.post_template.format(
proto=proto,
hostname_suffix=hostname_suffix,
asn=asn,
ip_type=ip_type,
+ location=location,
)
# Truncate to 300 char limit
def _send_follow_up(self, parent_ref, proto: str) -> None:
"""Post a follow-up reply to an announcement."""
try:
owner = self.config.owner_username.strip()
+ mention_text = f"@{owner}" if owner else ""
text = self.config.follow_up_template.format(
- owner_username=f"@{owner}" if owner else "",
+ owner_username=mention_text,
proto=proto,
)
if len(text) > 300:
text = text[:297] + "..."
+ # Build mention facet so @handle becomes a clickable mention
+ facets = []
+ if owner and mention_text in text:
+ try:
+ resolved = self.client.resolve_handle(owner)
+ mention_start = text.index(mention_text)
+ byte_start = len(text[:mention_start].encode("utf-8"))
+ byte_end = byte_start + len(mention_text.encode("utf-8"))
+ facets.append(models.AppBskyRichtextFacet.Main(
+ index=models.AppBskyRichtextFacet.ByteSlice(
+ byte_start=byte_start,
+ byte_end=byte_end,
+ ),
+ features=[models.AppBskyRichtextFacet.Mention(did=resolved.did)],
+ ))
+ except Exception as e:
+ logger.warning(f"Could not resolve handle {owner} for mention: {e}")
+
strong_ref = models.ComAtprotoRepoStrongRef.Main(
uri=parent_ref.uri,
cid=parent_ref.cid,
def _send_follow_up(self, parent_ref, proto: str) -> None:
root=strong_ref,
parent=strong_ref,
)
- self.client.send_post(text=text, reply_to=reply_ref)
+ self.client.send_post(text=text, reply_to=reply_ref, facets=facets or None)
logger.info("Posted follow-up reply to announcement")
except Exception as e:
logger.error(f"Failed to post follow-up reply: {e}")
File: backend/config.py
class AtprotoConfig:
username: str = ""
app_password: str = ""
owner_username: str = ""
- post_template: str = "Jackpot! Found an open {proto} host{hostname_suffix}\n{asn}\n{ip_type}"
+ post_template: str = "Jackpot! Found an open {proto} host{hostname_suffix}\n{location}\n{ip_type}"
follow_up_template: str = ""
+@dataclass
+class GeoipConfig:
+ enabled: bool = True
+ database_path: str = "geoip.db"
+ download_url_template: str = "https://download.db-ip.com/free/dbip-city-lite-{YYYY}-{MM}.csv.gz"
+
+
@dataclass
class Config:
app: AppConfig = field(default_factory=AppConfig)
scanner: ScannerConfig = field(default_factory=ScannerConfig)
atproto: AtprotoConfig = field(default_factory=AtprotoConfig)
+ geoip: GeoipConfig = field(default_factory=GeoipConfig)
def load_config(path: str = "config.toml") -> Config:
def load_config(path: str = "config.toml") -> Config:
app=AppConfig(**data.get("app", {})),
scanner=ScannerConfig(**data.get("scanner", {})),
atproto=AtprotoConfig(**data.get("atproto", {})),
+ geoip=GeoipConfig(**data.get("geoip", {})),
)
Read more...