Implement Domain Health Engine + Password Reset

🏥 DOMAIN HEALTH ENGINE (from analysis_2.md):
- New service: backend/app/services/domain_health.py
- 4-layer analysis:
  1. DNS: Nameservers, MX records, A records, parking NS detection
  2. HTTP: Status codes, content, parking keyword detection
  3. SSL: Certificate validity, expiration date, issuer
  4. (WHOIS via existing domain_checker)

📊 HEALTH SCORING:
- Score 0-100 based on all layers
- Status: HEALTHY (🟢), WEAKENING (🟡), PARKED (🟠), CRITICAL (🔴)
- Signals and recommendations for each domain

🔌 API ENDPOINTS:
- GET /api/v1/domains/{id}/health - Full health report
- POST /api/v1/domains/health-check?domain=x - Quick check any domain

🔐 PASSWORD RESET:
- New script: backend/scripts/reset_admin_password.py
- guggeryves@hotmail.com password: Pounce2024!

PARKING DETECTION:
- Known parking nameservers (Sedo, Afternic, etc.)
- Page content keywords ('buy this domain', 'for sale', etc.)
This commit is contained in:
yves.gugger
2025-12-10 09:34:43 +01:00
parent 1692da0519
commit a9b5cc0f82
4 changed files with 717 additions and 167 deletions

View File

@ -1,198 +1,112 @@
Hier ist die komplette **Master-Zusammenfassung** für `pounce.ch`. Dies ist dein Bauplan (Blueprint) für die Umsetzung.
Das ist der Kern deiner **"Intelligence Platform"**.
Wenn du keine externen APIs nutzt, baust du dir im Grunde einen **Gesundheits-Check für Domains**. Dein System fungiert als digitaler Arzt, der regelmäßig den Puls der Domain fühlt. Wenn der Puls schwächer wird (Webseite offline, Mails kommen zurück), alarmierst du deinen User.
Hier ist der technische und logische Ablauf, wie die **Pounce Domain-Analyse** (Engine) funktioniert.
Wir teilen die Analyse in **4 Ebenen (Layers)** auf:
---
### 1. Die Vision & Positionierung
**Name:** Pounce
**Tagline:** *Domain Intelligence for Hunters.*
**Slogan:** *"Don't guess. Know."*
**Konzept:** Pounce ist das "Bloomberg Terminal" für Domains. Es verwandelt den unübersichtlichen, lauten Domain-Markt in klare, handlungsfähige Daten. Es richtet sich an Leute, die nicht suchen, sondern finden wollen.
### Ebene 1: Der DNS-Check (Die Infrastruktur)
*Das ist der "Wohnsitz"-Check. Wohnt hier noch wer?*
* **Zielgruppe:**
* **Dreamers (Gründer):** Suchen den perfekten Namen für ihr Projekt.
* **Hunters (Investoren/Händler):** Suchen unterbewertete Assets für Arbitrage (günstig kaufen, teuer verkaufen).
Hier prüfst du die DNS-Einträge (Domain Name System). Das kostet dich fast keine Rechenleistung und geht extrem schnell.
**Was dein Skript prüft:**
1. **NS Records (Nameserver):** Wer verwaltet die Domain?
* *Signal:* Wechselt der Nameserver von `ns1.hostpoint.ch` (normales Hosting) zu `ns1.sedoparking.com` oder `ns1.afternic.com`?
* *Bedeutung:* **ALARM!** Der Besitzer hat das Projekt aufgegeben und die Domain zum Verkauf ("Parking") freigegeben. Das ist der beste Moment für ein Kaufangebot.
2. **A Record (IP-Adresse):** Zeigt die Domain auf einen Server?
* *Signal:* Eintrag wird gelöscht oder zeigt auf `0.0.0.0` oder `127.0.0.1`.
* *Bedeutung:* Die Domain ist "technisch tot". Sie löst nirgendwohin auf.
3. **MX Record (Mail Exchange):** Kann die Domain E-Mails empfangen?
* *Signal:* MX Records verschwinden.
* *Bedeutung:* Die Firma nutzt keine E-Mails mehr unter dieser Domain. Ein sehr starkes Zeichen für Geschäftsaufgabe.
---
### 2. Die 3 Produktsäulen (Das "Command Center")
### Ebene 2: Der HTTP-Check (Die Schaufenster-Analyse)
*Das ist der visuelle Check. Ist der Laden noch offen?*
Das Produkt gliedert sich logisch in drei Phasen der Domain-Beschaffung:
Hier versucht dein Bot, die Webseite tatsächlich aufzurufen (wie ein Browser, aber ohne Bilder zu laden).
#### A. DISCOVER (Markt-Intelligenz)
*Der "Honigtopf", um User anzuziehen (SEO & Traffic).*
* **TLD Intel:** Zeigt Markttrends (z.B. `.ai` steigt um 35%).
* **Smart Search:** Wenn eine Domain vergeben ist, zeigt Pounce **intelligente Alternativen** (z.B. `.io` für Tech, `.shop` für E-Commerce), statt nur zufällige Endungen.
* **Der Hook:** Öffentliche Besucher sehen Trends, aber Details (Charts, Historie) sind ausgeblendet ("Sign in to view").
#### B. TRACK (Die Watchlist)
*Das Tool für Kundenbindung.*
* **Funktion:** Überwachung von *vergebenen* Domains.
* **Der USP:** Nicht nur "frei/besetzt", sondern **"Pre-Drop Indicators"**. Warnung bei DNS-Änderungen oder wenn die Webseite offline geht. Das gibt dem User einen Zeitvorsprung vor der Konkurrenz.
#### C. ACQUIRE (Der Auktions-Aggregator)
*Der Hauptgrund für das Upgrade.*
* **Funktion:** Aggregiert Live-Auktionen von GoDaddy, Sedo, NameJet & DropCatch an einem Ort.
* **Der "Killer-Feature" (Spam-Filter):**
* *Free User:* Sieht alles (auch "Müll"-Domains wie `kredit-24-online.info`).
* *Paid User:* Sieht einen **kuratierten Feed**. Der Algorithmus filtert Zahlen, Bindestriche und Spam raus. Übrig bleiben nur hochwertige Investitions-Chancen.
**Was dein Skript prüft:**
1. **Status Codes (Der Türsteher):**
* **200 OK:** Seite ist online.
* **404 Not Found:** Seite existiert nicht (Datei fehlt).
* **500/503 Server Error:** Die Webseite ist kaputt.
* **Connection Refused / Timeout:** Der Server ist abgeschaltet.
* *Pounce Logic:* Ein Wechsel von **200** auf **Timeout** über 3 Tage hinweg ist ein starkes "Drop"-Signal.
2. **Content-Length (Größe der Seite):**
* *Signal:* Die Seite war früher 2MB groß, jetzt sind es nur noch 500 Bytes.
* *Bedeutung:* Der Inhalt wurde gelöscht, es steht nur noch "Coming Soon" oder eine weiße Seite da.
3. **Keyword-Scanning (Parked Detection):**
* Das Problem: Park-Seiten (Werbung) geben oft auch einen "200 OK" Status zurück.
* *Lösung:* Dein Skript scannt den HTML-Text nach Wörtern wie: *"Domain is for sale"*, *"Inquire now"*, *"Related Links"*, *"Buy this domain"*.
* *Bedeutung:* Wenn diese Wörter auftauchen, markierst du die Domain automatisch als **"On Sale / Parked"**.
---
### 3. Das Geschäftsmodell (Pricing)
### Ebene 3: Der SSL-Check (Die Wartung)
*Kümmert sich der Hausmeister noch?*
Das Modell basiert auf "Freemium mit Schranken". Der Preis von $9 ist ein "No-Brainer" (Impulskauf), um die Hürde niedrig zu halten.
Sicherheitszertifikate (SSL/TLS) müssen regelmäßig erneuert werden (oft alle 90 Tage bei Let's Encrypt, oder jährlich).
| Plan | Preis | Zielgruppe | Haupt-Features | Der "Schmerz" (Warum upgraden?) |
| :--- | :--- | :--- | :--- | :--- |
| **SCOUT** | **0 €** | Neugierige | 5 Watchlist-Domains, roher Auktions-Feed, Basis-Suche. | Muss sich durch "Spam" wühlen, sieht keine Bewertungen, langsame Alerts. |
| **TRADER** | **9 €** | Hobby-Investoren | 50 Watchlist-Domains, **Spam-freier Feed**, Deal Scores (Bewertungen), stündliche Checks. | Zahlt für Zeitersparnis (Filter) und Sicherheit (Bewertung). |
| **TYCOON** | **29 €** | Profis | 500 Domains, Echtzeit-Checks (10 Min), API-Zugriff (geplant). | Braucht Volumen und Geschwindigkeit. |
**Was dein Skript prüft:**
1. **Expiry Date des Zertifikats:**
* *Signal:* Das Zertifikat ist gestern abgelaufen ("Expired").
* *Bedeutung:* Der Admin kümmert sich nicht mehr. Moderne Browser zeigen jetzt eine Warnung ("Nicht sicher"). Besucher bleiben aus. Das Projekt stirbt.
---
### 4. UX/UI & Tone of Voice
### Ebene 4: Der Whois/RDAP Check (Der Vertrag)
*Wann läuft der Mietvertrag aus?*
* **Design-Philosophie:** "Dark Mode & Data".
* Dunkler Hintergrund (Schwarz/Grau) wirkt professionell (wie Trading-Software).
* Akzentfarben: Neon-Grün (für "Frei" / "Profit") und Warn-Orange.
* Wenig Text, viele Datenpunkte, klare Tabellen.
* **Tone of Voice:**
* Knapp, präzise, strategisch.
* Kein Marketing-Bla-Bla.
* *Beispiel:* Statt "Wir haben viele tolle Funktionen" → "Three moves to dominate."
Das ist der Check direkt bei der Registry (z.B. Verisign oder SWITCH). Da Whois oft Rate-Limits hat (du darfst nicht zu oft abfragen), machst du das seltener (z.B. 1x täglich). Nutze dafür am besten **RDAP** (Registration Data Access Protocol) das ist der moderne, maschinenlesbare Nachfolger von Whois (JSON Format).
**Was dein Skript prüft:**
1. **Expiration Date:** Wann läuft die Domain aus?
2. **Domain Status Codes (EPP Codes):**
* `clientTransferProhibited`: Alles normal (gesperrt gegen Diebstahl).
* `clientHold` oder `serverHold`: **JACKPOT!** Die Domain wurde deaktiviert (meist wegen Nichtzahlung). Sie wird sehr bald gelöscht.
* `redemptionPeriod`: Die Gnadenfrist läuft. Der Besitzer muss Strafe zahlen, um sie zu retten. Tut er es nicht, droppt sie in ~30 Tagen.
---
### 5. Die User Journey (Der "Golden Path")
### Zusammenfassung: Der "Pounce Health Score"
1. **Der Einstieg:** User googelt "Domain Preise .ai" und landet auf deiner **TLD Intel Page**.
2. **Der Hook:** Er sieht "`.ai` +35%", will aber die Details sehen. Die Tabelle ist unscharf. Button: *"Sign In to view details"*.
3. **Die Registrierung:** Er erstellt einen Free Account ("Scout").
4. **Die Erkenntnis:** Er geht zu den Auktionen. Er sieht eine interessante Domain, aber weiß nicht, ob der Preis gut ist. Neben dem Preis steht: *"Valuation locked"*.
5. **Das Upgrade:** Er sieht das Angebot: "Für nur $9/Monat siehst du den echten Wert und wir filtern den Müll für dich."
6. **Der Kauf:** Er abonniert den "Trader"-Plan.
Damit der User nicht mit technischen Daten erschlagen wird, fasst du all diese Checks in einem einfachen Status im Dashboard zusammen.
---
**Beispiel-Logik für deine App:**
### Zusammenfassung für den Entwickler (Tech Stack Requirements)
* **Status: 🟢 HEALTHY (Aktiv)**
* DNS: OK
* HTTP: 200 OK
* SSL: Valid
* **Frontend:** Muss extrem schnell sein (Reagierende Suche). Mobile-freundlich (Tabellen müssen auf dem Handy lesbar sein oder ausgeblendet werden).
* **Daten-Integration:** APIs zu GoDaddy, Sedo etc. oder Scraping für die Auktionsdaten.
* **Logik:**
* **Filter-Algorithmus:** Das Wichtigste! (Regeln: Keine Zahlen, max. 2 Bindestriche, Wörterbuch-Abgleich).
* **Alert-System:** Cronjobs für E-Mail/SMS Benachrichtigungen.
* **Status: 🟡 WEAKENING (Schwächelnd - Watchlist Alarm!)**
* SSL: Expired ⚠️
* HTTP: 500 Error oder Content-Length drastisch gesunken ⚠️
* *Nachricht an User:* "Webseite ist kaputt gegangen und Zertifikat abgelaufen. Besitzer verliert Interesse."
Das Konzept ist jetzt rund, logisch und bereit für den Bau. Viel Erfolg mit **Pounce**! 🚀
* **Status: 🟠 PARKED (Zu Verkaufen)**
* DNS: Zeigt auf Sedo/Afternic
* HTTP Body: Enthält "Buy this domain"
Das ist eine sehr kluge Entscheidung für den Start (Lean Startup). Externe APIs (wie GoDaddy Auction API, Estibot etc.) sind oft teuer, komplex in der Anbindung oder erfordern hohe Händler-Umsätze.
* **Status: 🔴 CRITICAL / PENDING DROP (Gleich weg)**
* Whois Status: `redemptionPeriod` oder `clientHold`
* DNS: NXDOMAIN (Existiert nicht mehr)
* *Nachricht an User:* "Domain wurde vom Registrar deaktiviert. Drop steht bevor!"
Die gute Nachricht: Du kannst eine **High-End-Plattform** bauen, die komplett auf **eigenen Daten und Skripten** basiert. Das macht dich unabhängig und deine Daten *einzigartig* (weil du nicht dasselbe anzeigst wie alle anderen).
### Technische Umsetzung (Tech Stack für Python)
Hier ist deine Strategie für das **"No-API, High-Value"** Konzept:
Wenn du das bauen willst, brauchst du folgende Python-Libraries (alle Open Source):
-----
1. **DNS:** `dnspython` (um Nameserver und MX Records abzufragen).
2. **HTTP:** `requests` (um Status Codes und Content zu prüfen).
3. **SSL:** `ssl` & `socket` (Standard-Libraries, um Zertifikatsdatum auszulesen).
4. **Whois:** `python-whois` (einfacher Wrapper) oder direkte RDAP-Abfragen via `requests`.
### Die Geheimwaffe: Zone Files & DNS
Statt Daten einzukaufen, generierst du sie selbst aus der Rohmasse des Internets.
#### 1\. Die Datenquelle: TLD Zone Files (Der Rohstoff)
Jede Registry (Verisign für .com, SWITCH für .ch/.li) führt ein sogenanntes **Zone File**. Das ist eine riesige Textdatei mit *allen* aktiven Domains dieser Endung.
* **Kosten:** Oft kostenlos oder gegen kleine Gebühr beantragbar (bei Verisign z.B. Zugriff via CZDS - Centralized Zone Data Service).
* **Deine Methode:**
1. Du lädst jeden Tag das Zone File für .com, .net, .ch, .io herunter.
2. Du vergleichst die Datei von **Heute** mit der von **Gestern**.
3. **Resultat:** Du weißt exakt, welche Domains **gelöscht (Deleted)** wurden und welche **neu registriert (Added)** wurden.
Das ist Gold wert. Du hast damit deine eigene Datenbank an "Just Dropped" Domains, ohne jemanden zu fragen.
-----
### Das Produkt-Angebot (Ohne externe APIs)
So baust du die Features auf, damit sie "Premium" wirken:
#### Feature A: "The Daily Drop List" (Statt Live-Auktionen)
Da du keine Live-Auktionsdaten von GoDaddy hast, fokussierst du dich auf **"Pending Deletes"** und **"Fresh Drops"**.
* **Was du tust:** Dein Skript nimmt die Liste der 50.000 Domains, die gestern gelöscht wurden.
* **Der Premium-Faktor (Dein Algorithmus):** Du jagst diese Liste durch eigene Python-Filter:
* *Dictionary Check:* Ist der Domainname in einem englischen/deutschen Wörterbuch?
* *Pattern Check:* Ist es CVCV (Consonant-Vowel... z.B. "bamo.com")?
* *Length Check:* Kürzer als 5 Zeichen?
* **Das User-Versprechen:** "Täglich löschen 100.000 Domains. Wir zeigen dir die 50, die etwas wert sind."
* **Warum User zahlen:** Sie sparen Zeit. Die Rohdaten sind nutzlos, deine gefilterte Liste ist Geld.
#### Feature B: "Smart Monitor" (Eigene Status-Checks)
Du brauchst keine API, um zu wissen, ob eine Webseite offline ist.
* **Technik:** Ein einfacher Cronjob auf deinem Server.
* **HTTP-Check:** Sende eine Anfrage an `meinedomain.ch`. Kommt ein Fehler (404, 500) oder "Connection Refused"? -\> **Alarm\!**
* **DNS-Check:** Ändern sich die Nameserver (z.B. von "https://www.google.com/search?q=ns.wix.com" auf "https://www.google.com/search?q=ns.sedoparking.com")? -\> **Alarm\!**
* **Premium Feature:** "Der Silent-Alarm". Der User gibt Domains ein. Dein Server prüft sie alle 6 Stunden. Wenn sich der Status ändert, geht eine E-Mail raus.
* **Kosten für dich:** Nahezu Null (nur Server-Last).
#### Feature C: "Pounce Internal Score" (Statt Estibot)
Du kannst keine externe Bewertung (Valuation) abrufen. Also erfindest du deinen eigenen **Score**.
* **Die Logik:** Du vergibst Punkte basierend auf objektiven Kriterien:
* Länge \< 5 Zeichen: +50 Punkte
* Endung ist .com/.ch: +30 Punkte
* Keine Zahlen/Bindestriche: +20 Punkte
* Wörterbuch-Treffer: +100 Punkte
* **Marketing:** Nenne es **"Pounce Opportunity Score"**. Das wirkt exklusiv. User lieben Scores, egal wie simpel die Mathematik dahinter ist, solange sie hilft, Schrott auszusortieren.
#### Feature D: TLD Trends (Deine eigenen Charts)
Da du die Zone Files hast, kannst du die Charts selbst zeichnen.
* Zähle einfach jeden Tag die Zeilen im .ai Zone File.
* Tag 1: 150.000 Domains. Tag 2: 150.500 Domains.
* **Anzeige:** "Wachstum .ai: +0.3% heute".
* Das sind **echte, proprietäre Daten**, die sonst niemand so einfach hat.
-----
### Die neue Informationsarchitektur (Lean Version)
Die Landing Page bleibt fast gleich, aber das Wording ändert sich leicht, um "fehlende" Live-Auktionen zu kaschieren.
**1. Discover (Drops)**
Statt "Live Auctions" nennst du es **"Daily Drop Gems"**.
* *Versprechen:* "Jeden Morgen um 08:00 Uhr: Die Top 100 gelöschten Domains des Tages."
* *Quelle:* Deine Zone-File-Analyse.
**2. Track (Monitor)**
Statt "Real-time Registry Scan" nennst du es **"Change Detection"**.
* *Versprechen:* "Wir pingen deine Wunsch-Domains 4x täglich. Wenn sie offline gehen, weißt du es sofort."
**3. Intelligence (Trends)**
Das bleibt gleich. Deine eigenen Charts basierend auf den Zone-Countings.
-----
### Wie du das technisch startest (Roadmap)
1. **Schritt 1 (Access):** Melde dich beim **CZDS (ICANN)** an, um Zugriff auf die Zone Files (.com, .net etc.) zu bekommen. Für .ch/.li schau bei SWITCH nach Open Data oder nutze Listen von Anbietern wie "daily-domain-lists" (die kosten oft nur $5/Monat für Rohdaten).
2. **Schritt 2 (Scripting):** Schreibe ein Python-Skript, das:
* Die Files downloaded.
* Diffs erstellt (Was ist weg?).
* Die gelöschten Domains durch einen Wörterbuch-Filter jagt.
3. **Schritt 3 (Frontend):** Zeige diese gefilterte Liste in deinem Dashboard an.
4. **Schritt 4 (Pricing):**
* Free: Sieht die Liste von *gestern*.
* Paid ($9): Sieht die Liste von *heute* + "Pounce Score".
### Fazit
Du brauchst keine APIs.
Tatsächlich ist der Ansatz **"Zone File Analysis + Eigener Filter"** oft besser, weil du **unabhängig** bist. Du baust eine echte Daten-Company auf, keinen bloßen API-Wrapper (der nur Daten von A nach B schiebt).
Das ist **echter Mehrwert**. Und dafür zahlen Leute.
**Pro-Tipp für deinen Server:**
Da du viele Domains checkst, darfst du das nicht "hintereinander" machen (dauert zu lange). Du musst es **asynchron** machen (viele gleichzeitig). Schau dir dafür **Python `asyncio`** und **`aiohttp`** an. Damit kannst du Tausende Domains in wenigen Minuten prüfen.

View File

@ -11,6 +11,7 @@ from app.models.domain import Domain, DomainCheck, DomainStatus
from app.models.subscription import TIER_CONFIG, SubscriptionTier
from app.schemas.domain import DomainCreate, DomainResponse, DomainListResponse
from app.services.domain_checker import domain_checker
from app.services.domain_health import get_health_checker, HealthStatus
router = APIRouter()
@ -312,3 +313,60 @@ async def get_domain_history(
]
}
@router.get("/{domain_id}/health")
async def get_domain_health(
domain_id: int,
current_user: CurrentUser,
db: Database,
):
"""
Get comprehensive health report for a domain.
Checks 4 layers:
- DNS: Nameservers, MX records, A records
- HTTP: Website availability, parking detection
- SSL: Certificate validity and expiration
- Status signals and recommendations
Returns:
Health report with score (0-100) and status
"""
# Get domain
result = await db.execute(
select(Domain).where(
Domain.id == domain_id,
Domain.user_id == current_user.id,
)
)
domain = result.scalar_one_or_none()
if not domain:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Domain not found",
)
# Run health check
health_checker = get_health_checker()
report = await health_checker.check_domain(domain.name)
return report.to_dict()
@router.post("/health-check")
async def quick_health_check(
current_user: CurrentUser,
domain: str = Query(..., description="Domain to check"),
):
"""
Quick health check for any domain (doesn't need to be in watchlist).
Premium feature - checks DNS, HTTP, and SSL layers.
"""
# Run health check
health_checker = get_health_checker()
report = await health_checker.check_domain(domain)
return report.to_dict()

View File

@ -0,0 +1,521 @@
"""
🏥 POUNCE DOMAIN HEALTH ENGINE
Advanced domain health analysis for premium intelligence.
Implements 4-layer analysis from analysis_2.md:
1. DNS Layer - Infrastructure check (nameservers, MX, A records)
2. HTTP Layer - Website availability (status codes, content, parking detection)
3. SSL Layer - Certificate validity
4. WHOIS/RDAP Layer - Registration status
Output: Health Score (HEALTHY, WEAKENING, PARKED, CRITICAL)
"""
import asyncio
import logging
import ssl
import socket
import re
from datetime import datetime, timezone, timedelta
from dataclasses import dataclass, field
from typing import Optional, List, Dict, Any
from enum import Enum
import httpx
import dns.resolver
import dns.exception
logger = logging.getLogger(__name__)
class HealthStatus(str, Enum):
"""Domain health status levels."""
HEALTHY = "healthy" # 🟢 All systems go
WEAKENING = "weakening" # 🟡 Warning signs detected
PARKED = "parked" # 🟠 Domain for sale/parked
CRITICAL = "critical" # 🔴 Drop imminent
UNKNOWN = "unknown" # ❓ Could not determine
@dataclass
class DNSCheckResult:
"""Results from DNS layer check."""
has_nameservers: bool = False
nameservers: List[str] = field(default_factory=list)
has_mx_records: bool = False
mx_records: List[str] = field(default_factory=list)
has_a_record: bool = False
a_records: List[str] = field(default_factory=list)
is_parking_ns: bool = False # Nameservers point to parking service
error: Optional[str] = None
@dataclass
class HTTPCheckResult:
"""Results from HTTP layer check."""
status_code: Optional[int] = None
is_reachable: bool = False
content_length: int = 0
is_parked: bool = False
parking_signals: List[str] = field(default_factory=list)
redirect_url: Optional[str] = None
response_time_ms: Optional[float] = None
error: Optional[str] = None
@dataclass
class SSLCheckResult:
"""Results from SSL layer check."""
has_ssl: bool = False
is_valid: bool = False
expires_at: Optional[datetime] = None
days_until_expiry: Optional[int] = None
issuer: Optional[str] = None
is_expired: bool = False
error: Optional[str] = None
@dataclass
class DomainHealthReport:
"""Complete health report for a domain."""
domain: str
status: HealthStatus
score: int # 0-100
# Layer results
dns: Optional[DNSCheckResult] = None
http: Optional[HTTPCheckResult] = None
ssl: Optional[SSLCheckResult] = None
# Summary
signals: List[str] = field(default_factory=list)
recommendations: List[str] = field(default_factory=list)
# Metadata
checked_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary for API response."""
return {
"domain": self.domain,
"status": self.status.value,
"score": self.score,
"signals": self.signals,
"recommendations": self.recommendations,
"checked_at": self.checked_at.isoformat(),
"layers": {
"dns": {
"has_nameservers": self.dns.has_nameservers if self.dns else False,
"nameservers": self.dns.nameservers if self.dns else [],
"has_mx_records": self.dns.has_mx_records if self.dns else False,
"is_parking_ns": self.dns.is_parking_ns if self.dns else False,
} if self.dns else None,
"http": {
"status_code": self.http.status_code if self.http else None,
"is_reachable": self.http.is_reachable if self.http else False,
"is_parked": self.http.is_parked if self.http else False,
"response_time_ms": self.http.response_time_ms if self.http else None,
} if self.http else None,
"ssl": {
"has_ssl": self.ssl.has_ssl if self.ssl else False,
"is_valid": self.ssl.is_valid if self.ssl else False,
"days_until_expiry": self.ssl.days_until_expiry if self.ssl else None,
"is_expired": self.ssl.is_expired if self.ssl else False,
} if self.ssl else None,
}
}
class DomainHealthChecker:
"""
Premium domain health analysis engine.
Checks 4 layers to determine domain health:
1. DNS: Is the infrastructure alive?
2. HTTP: Is the website running?
3. SSL: Is the certificate valid?
4. (WHOIS handled by existing DomainChecker)
"""
# Known parking/for-sale service nameservers
PARKING_NAMESERVERS = {
'sedoparking.com', 'afternic.com', 'domaincontrol.com',
'parkingcrew.net', 'bodis.com', 'dsredirection.com',
'above.com', 'domainsponsor.com', 'fastpark.net',
'parkdomain.com', 'domainmarket.com', 'hugedomains.com',
}
# Keywords indicating parked/for-sale pages
PARKING_KEYWORDS = [
'domain is for sale', 'buy this domain', 'inquire now',
'make an offer', 'domain zum verkauf', 'domain for sale',
'this domain is parked', 'parked by', 'related links',
'sponsored listings', 'domain parking', 'this website is for sale',
'purchase this domain', 'acquire this domain',
]
def __init__(self):
self._dns_resolver = dns.resolver.Resolver()
self._dns_resolver.timeout = 3
self._dns_resolver.lifetime = 5
async def check_domain(self, domain: str) -> DomainHealthReport:
"""
Perform comprehensive health check on a domain.
Args:
domain: Domain name to check (e.g., "example.com")
Returns:
DomainHealthReport with status, score, and detailed results
"""
domain = self._normalize_domain(domain)
logger.info(f"🏥 Starting health check for: {domain}")
# Run all checks concurrently
dns_task = asyncio.create_task(self._check_dns(domain))
http_task = asyncio.create_task(self._check_http(domain))
ssl_task = asyncio.create_task(self._check_ssl(domain))
dns_result, http_result, ssl_result = await asyncio.gather(
dns_task, http_task, ssl_task,
return_exceptions=True
)
# Handle exceptions
if isinstance(dns_result, Exception):
logger.warning(f"DNS check failed: {dns_result}")
dns_result = DNSCheckResult(error=str(dns_result))
if isinstance(http_result, Exception):
logger.warning(f"HTTP check failed: {http_result}")
http_result = HTTPCheckResult(error=str(http_result))
if isinstance(ssl_result, Exception):
logger.warning(f"SSL check failed: {ssl_result}")
ssl_result = SSLCheckResult(error=str(ssl_result))
# Calculate health score and status
report = self._calculate_health(domain, dns_result, http_result, ssl_result)
logger.info(f"✅ Health check complete: {domain} = {report.status.value} ({report.score}/100)")
return report
def _normalize_domain(self, domain: str) -> str:
"""Normalize domain name."""
domain = domain.lower().strip()
if domain.startswith('http://'):
domain = domain[7:]
elif domain.startswith('https://'):
domain = domain[8:]
if domain.startswith('www.'):
domain = domain[4:]
domain = domain.split('/')[0]
return domain
async def _check_dns(self, domain: str) -> DNSCheckResult:
"""
Layer 1: DNS Infrastructure Check
Checks:
- NS records (nameservers)
- MX records (mail)
- A records (IP address)
"""
result = DNSCheckResult()
loop = asyncio.get_event_loop()
# Check NS records
try:
ns_answers = await loop.run_in_executor(
None, lambda: self._dns_resolver.resolve(domain, 'NS')
)
result.nameservers = [str(rdata.target).rstrip('.').lower() for rdata in ns_answers]
result.has_nameservers = len(result.nameservers) > 0
# Check if nameservers point to parking service
for ns in result.nameservers:
for parking_ns in self.PARKING_NAMESERVERS:
if parking_ns in ns:
result.is_parking_ns = True
break
except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer, dns.exception.Timeout):
result.has_nameservers = False
except Exception as e:
result.error = str(e)
# Check MX records
try:
mx_answers = await loop.run_in_executor(
None, lambda: self._dns_resolver.resolve(domain, 'MX')
)
result.mx_records = [str(rdata.exchange).rstrip('.').lower() for rdata in mx_answers]
result.has_mx_records = len(result.mx_records) > 0
except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer, dns.exception.Timeout):
result.has_mx_records = False
except Exception:
pass
# Check A records
try:
a_answers = await loop.run_in_executor(
None, lambda: self._dns_resolver.resolve(domain, 'A')
)
result.a_records = [str(rdata.address) for rdata in a_answers]
result.has_a_record = len(result.a_records) > 0
# Check for dead IPs (0.0.0.0 or 127.0.0.1)
dead_ips = {'0.0.0.0', '127.0.0.1'}
if all(ip in dead_ips for ip in result.a_records):
result.has_a_record = False
except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer, dns.exception.Timeout):
result.has_a_record = False
except Exception:
pass
return result
async def _check_http(self, domain: str) -> HTTPCheckResult:
"""
Layer 2: HTTP Website Check
Checks:
- HTTP status code
- Response content
- Parking/for-sale detection
"""
result = HTTPCheckResult()
async with httpx.AsyncClient(
timeout=10.0,
follow_redirects=True,
headers={
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}
) as client:
for scheme in ['https', 'http']:
url = f"{scheme}://{domain}"
try:
start = asyncio.get_event_loop().time()
response = await client.get(url)
end = asyncio.get_event_loop().time()
result.status_code = response.status_code
result.is_reachable = response.status_code < 500
result.content_length = len(response.content)
result.response_time_ms = (end - start) * 1000
# Check for redirects
if response.history:
result.redirect_url = str(response.url)
# Check for parking keywords in content
content = response.text.lower()
for keyword in self.PARKING_KEYWORDS:
if keyword in content:
result.is_parked = True
result.parking_signals.append(keyword)
break # Success, no need to try other scheme
except httpx.TimeoutException:
result.error = "timeout"
except httpx.ConnectError:
result.error = "connection_refused"
except Exception as e:
result.error = str(e)
return result
async def _check_ssl(self, domain: str) -> SSLCheckResult:
"""
Layer 3: SSL Certificate Check
Checks:
- Certificate exists
- Certificate validity
- Expiration date
"""
result = SSLCheckResult()
loop = asyncio.get_event_loop()
try:
def get_ssl_info():
context = ssl.create_default_context()
with socket.create_connection((domain, 443), timeout=5) as sock:
with context.wrap_socket(sock, server_hostname=domain) as ssock:
cert = ssock.getpeercert()
return cert
cert = await loop.run_in_executor(None, get_ssl_info)
result.has_ssl = True
# Parse expiration date
not_after = cert.get('notAfter')
if not_after:
# Format: 'Dec 31 23:59:59 2024 GMT'
try:
expires = datetime.strptime(not_after, '%b %d %H:%M:%S %Y %Z')
result.expires_at = expires.replace(tzinfo=timezone.utc)
result.days_until_expiry = (result.expires_at - datetime.now(timezone.utc)).days
result.is_expired = result.days_until_expiry < 0
result.is_valid = result.days_until_expiry >= 0
except Exception:
result.is_valid = True # Assume valid if we can't parse
# Get issuer
issuer = cert.get('issuer')
if issuer:
for item in issuer:
if item[0][0] == 'organizationName':
result.issuer = item[0][1]
break
except ssl.SSLCertVerificationError as e:
result.has_ssl = True
result.is_valid = False
result.is_expired = 'expired' in str(e).lower()
result.error = str(e)
except (socket.timeout, socket.error, ConnectionRefusedError):
result.has_ssl = False
result.error = "no_ssl"
except Exception as e:
result.error = str(e)
return result
def _calculate_health(
self,
domain: str,
dns_result: DNSCheckResult,
http_result: HTTPCheckResult,
ssl_result: SSLCheckResult
) -> DomainHealthReport:
"""
Calculate overall health status and score.
Scoring:
- DNS layer: 30 points
- HTTP layer: 40 points
- SSL layer: 30 points
"""
score = 100
signals = []
recommendations = []
# =========================
# DNS Scoring (30 points)
# =========================
if not dns_result.has_nameservers:
score -= 30
signals.append("🔴 No nameservers found (domain may not exist)")
elif dns_result.is_parking_ns:
score -= 15
signals.append("🟠 Nameservers point to parking service")
recommendations.append("Domain is parked - owner may be selling")
else:
if not dns_result.has_a_record:
score -= 10
signals.append("⚠️ No A record (no website configured)")
if not dns_result.has_mx_records:
score -= 5
signals.append("⚠️ No MX records (no email configured)")
# =========================
# HTTP Scoring (40 points)
# =========================
if not http_result.is_reachable:
score -= 40
signals.append("🔴 Website not reachable")
if http_result.error == "timeout":
signals.append("⚠️ Connection timeout")
elif http_result.error == "connection_refused":
signals.append("⚠️ Connection refused")
elif http_result.status_code:
if http_result.status_code >= 500:
score -= 30
signals.append(f"🔴 Server error ({http_result.status_code})")
recommendations.append("Server is having issues - monitor closely")
elif http_result.status_code >= 400:
score -= 15
signals.append(f"⚠️ Client error ({http_result.status_code})")
if http_result.is_parked:
score -= 10
signals.append("🟠 Page contains for-sale indicators")
recommendations.append(f"Detected: {', '.join(http_result.parking_signals[:3])}")
if http_result.content_length < 500:
score -= 5
signals.append("⚠️ Very small page content")
# =========================
# SSL Scoring (30 points)
# =========================
if not ssl_result.has_ssl:
score -= 10
signals.append("⚠️ No SSL certificate")
elif ssl_result.is_expired:
score -= 30
signals.append("🔴 SSL certificate expired!")
recommendations.append("Certificate expired - owner neglecting domain")
elif ssl_result.days_until_expiry is not None:
if ssl_result.days_until_expiry < 7:
score -= 15
signals.append(f"⚠️ SSL expires in {ssl_result.days_until_expiry} days")
recommendations.append("Certificate expiring soon - watch for neglect")
elif ssl_result.days_until_expiry < 30:
score -= 5
signals.append(f" SSL expires in {ssl_result.days_until_expiry} days")
# Ensure score is in valid range
score = max(0, min(100, score))
# Determine status
if score >= 80:
status = HealthStatus.HEALTHY
elif score >= 50:
if dns_result.is_parking_ns or http_result.is_parked:
status = HealthStatus.PARKED
else:
status = HealthStatus.WEAKENING
elif score >= 20:
if dns_result.is_parking_ns or http_result.is_parked:
status = HealthStatus.PARKED
else:
status = HealthStatus.WEAKENING
else:
status = HealthStatus.CRITICAL
# Override to PARKED if clear signals
if dns_result.is_parking_ns or http_result.is_parked:
if status != HealthStatus.CRITICAL:
status = HealthStatus.PARKED
return DomainHealthReport(
domain=domain,
status=status,
score=score,
dns=dns_result,
http=http_result,
ssl=ssl_result,
signals=signals,
recommendations=recommendations,
)
# Singleton instance
_health_checker: Optional[DomainHealthChecker] = None
def get_health_checker() -> DomainHealthChecker:
"""Get or create health checker instance."""
global _health_checker
if _health_checker is None:
_health_checker = DomainHealthChecker()
return _health_checker

View File

@ -0,0 +1,57 @@
#!/usr/bin/env python3
"""
Script to reset admin password.
"""
import asyncio
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
from sqlalchemy import select
from passlib.context import CryptContext
from app.database import AsyncSessionLocal
from app.models.user import User
ADMIN_EMAIL = "guggeryves@hotmail.com"
NEW_PASSWORD = "Pounce2024!" # Strong password
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
async def reset_password():
"""Reset admin password."""
print(f"🔐 Resetting password for: {ADMIN_EMAIL}")
async with AsyncSessionLocal() as db:
result = await db.execute(
select(User).where(User.email == ADMIN_EMAIL)
)
user = result.scalar_one_or_none()
if not user:
print(f"❌ User not found: {ADMIN_EMAIL}")
return False
# Hash new password
hashed = pwd_context.hash(NEW_PASSWORD)
user.hashed_password = hashed
user.is_verified = True
user.is_active = True
user.is_admin = True
await db.commit()
print(f"✅ Password reset successful!")
print(f"\n📋 LOGIN CREDENTIALS:")
print(f" Email: {ADMIN_EMAIL}")
print(f" Password: {NEW_PASSWORD}")
print(f"\n⚠️ Please change this password after logging in!")
return True
if __name__ == "__main__":
asyncio.run(reset_password())