Refactor: migrate all providers to BaseProvider ABC

eagleosint/plugin.py

@@ -2,7 +2,10 @@
 BaseProvider ABC and ProviderCategory Enum
 
 Every provider plugin must subclass BaseProvider and implement execute().
-Existing CLI are not yet migrated -- this is the target interface.
+Migrated: GitHubProvider, BitlyProvider, PhoneInfoProvider,
+          IPLocationProvider, DomainInfoProvider, MailFinderProvider,
+          UserReconProvider, GoDorkerProvider.
+Pending:  Facebook (deprecated API — isolate/remove), TempMail (polling pattern).
 """
 
 from __future__ import annotations

eagleosint/providers/bitly.py

@@ -1,5 +1,8 @@
 """Bitly URL shortener bypass."""
+from __future__ import annotations
+import logging
 import urllib.parse
+from typing import ClassVar
 from getpass import getpass
 
 import requests
@@ -10,41 +13,72 @@
     BG_BLUE,
     SPACE_PREFIX, LINES_SEPARATOR,
 )
+from eagleosint.plugin import BaseProvider, ProviderCategory
 from eagleosint.session import session as _session
 from eagleosint.models import URLExpansion
 
+logger = logging.getLogger(__name__)
+
+class BitlyProvider(BaseProvider):
+    name = "bitly"
+    version = "1.0.0"
+    description = "Resolve shortened Bitly URLs to their original destination"
+    category = ProviderCategory.UTILITY
+    required_keys: ClassVar[list[str]] = []
+
+    def validate_query(self, query: str) -> str | None:
+        base = super().validate_query(query)
+        if base:
+            return base
+        parsed = urllib.parse.urlparse(query)
+        if parsed.scheme not in ("http", "https") or not parsed.netloc:
+            return "Invalid URL. Must start with http:// or https://"
+        return None
+
+    def execute(self, query: str) -> list[URLExpansion]:
+        error = self.validate_query(query)
+        if error:
+            logger.warning("Invalid query: %s", error)
+            return []
+
+        try:
+            response = _session.get(query, allow_redirects=False, timeout=10)
+            soup = BeautifulSoup(response.text, "lxml")
+            original_url = str(soup.find_all("a", href=True)[0]["href"])
+        except requests.exceptions.RequestException as e:
+            logger.error("Bitly fetch error: %s", e)
+            return []
+        except (IndexError, KeyError) as e:
+            logger.error("Bitly parse error: %s", e)
+            return []
+
+        return [URLExpansion(
+            query=query,
+            short_url=query,
+            original_url=original_url,
+        )]
 
 def bypass_bitly() -> URLExpansion | None:
-    """Bypasses Bitly URL shorteners."""
+    """Interactive CLI wrapper — prompts for URL, prints result."""
     print(WHITE + LINES_SEPARATOR)
-    bitly_url_input = input(
+    bitly_url = input(
         f"{SPACE_PREFIX}{WHITE}{BLUE}>{WHITE} Bitly URL: {BLUE}"
     ).strip()
-    parsed = urllib.parse.urlparse(bitly_url_input)
-    if parsed.scheme not in ("http", "https") or not parsed.netloc:
-        print(f"{RED}Invalid URL. Must start with http:// or https://{WHITE}")
-        return
-    try:
-        bitly_code_response = _session.get(
-            bitly_url_input, allow_redirects=False, timeout=10
-        )
-        soup_parser = BeautifulSoup(bitly_code_response.text, "lxml")
-        original_link_found = soup_parser.find_all("a", href=True)[0]["href"]
-        result = URLExpansion(
-            query=bitly_url_input,
-            short_url=bitly_url_input,
-            original_url=original_link_found,
-        )
-        print(
-            f"{SPACE_PREFIX}{BG_BLUE} DONE {WHITE} Original URL: "
-            f"\u001b[38;5;32m{original_link_found}"
-        )
-    except requests.exceptions.RequestException as e:
-        print(f"{RED}Error fetching Bitly URL: {e}{WHITE}")
-        return None
-    except (IndexError, KeyError) as e:
-        print(f"{RED}Error parsing Bitly response: {e}{WHITE}")
+
+    provider = BitlyProvider()
+    results = provider.execute(bitly_url)
+
+    if not results:
+        print(f"{RED}No results for '{bitly_url}'.{WHITE}")
+        print(WHITE + LINES_SEPARATOR)
+        getpass(SPACE_PREFIX + "press enter for back to previous menu ")
         return None
+
+    result = results[0]
+    print(
+        f"{SPACE_PREFIX}{BG_BLUE} DONE {WHITE} Original URL: "
+        f"\u001b[38;5;32m{result.original_url}"
+    )
     print(WHITE + LINES_SEPARATOR)
     getpass(SPACE_PREFIX + "press enter for back to previous menu ")
     return result

eagleosint/providers/github.py

@@ -2,6 +2,8 @@
 import json
 import re
 from getpass import getpass
+from typing import ClassVar
+import logging
 
 import requests
 from tabulate import tabulate
@@ -10,55 +12,86 @@
     RED, BLUE, WHITE,
     SPACE_PREFIX, LINES_SEPARATOR,
 )
+from eagleosint.plugin import BaseProvider, ProviderCategory
 from eagleosint.session import session as _session
 from eagleosint.models import GitHubProfile
 
+logger = logging.getLogger(__name__)
+
 GITHUB_API_URL = "https://api.github.com/users/{}"
 
+class GithubProvider(BaseProvider):
+    name = "github"
+    version = "1.0.0"
+    description = "GitHub user profile lookup via public REST API"
+    category = ProviderCategory.SOCIAL
+    required_keys: ClassVar[list[str]] = []
+
+    def execute(self, query: str) -> list[GitHubProfile]:
+        error = self.validate_query(query)
+        if error:
+            logger.warning("Invalid query: %s", error)
+            return []
+
+        if not re.match(r"^[a-zA-Z0-9][a-zA-Z0-9-]{0,37}$", query):
+            logger.warning("Invalid GitHub username: %s", query)
+            return []
+
+        try:
+            response = _session.get(GITHUB_API_URL.format(query), timeout=10)
+            response.raise_for_status()
+            data = response.json()
+        except requests.exceptions.RequestException as e:
+            logger.error("GitHub API error: %s", e)
+            return []
+        except json.JSONDecodeError:
+            logger.error("Invalid JSON response from GitHub API")
+            return []
+
+        profile = GitHubProfile(
+            query=query,
+            username=data.get("login", query),
+            name=data.get("name"),
+            email=data.get("email"),
+            bio=data.get("bio"),
+            location=data.get("location"),
+            company=data.get("company"),
+            blog=data.get("blog"),
+            public_repos=data.get("public_repos"),
+            followers=data.get("followers"),
+            following=data.get("following"),
+            created_at=data.get("created_at"),
+            avatar_url=data.get("avatar_url"),
+            html_url=data.get("html_url"),
+            raw=data,
+        )
+        return [profile]
 
 def github_lookup() -> GitHubProfile | None:
-    """Retrieves and displays information about a GitHub user."""
+    """Interactive CLI wrapper — prompts for username, prints table."""
     print(WHITE + LINES_SEPARATOR)
     github_user = input(
         f"{SPACE_PREFIX}{WHITE}{BLUE}>{WHITE} Github username: {BLUE}"
     ).strip()
-    if not re.match(r"^[a-zA-Z0-9][a-zA-Z0-9-]{0,37}$", github_user):
-        print(f"{RED}Invalid Github username.{WHITE}")
-        return
     print(WHITE)
-    try:
-        req = _session.get(GITHUB_API_URL.format(github_user), timeout=10)
-        res_data = req.json()
-        table_data = [[str(k), str(v)] for k, v in res_data.items()]
-        for line_item in tabulate(
-            table_data, headers=["info", "content"], tablefmt="fancy_grid"
-        ).splitlines():
-            print(" " * int(len(SPACE_PREFIX) / 2) + line_item)
 
-        profile = GitHubProfile(
-            query=github_user,
-            username=res_data.get("login", github_user),
-            name=res_data.get("name"),
-            email=res_data.get("email"),
-            bio=res_data.get("bio"),
-            location=res_data.get("location"),
-            company=res_data.get("company"),
-            blog=res_data.get("blog"),
-            public_repos=res_data.get("public_repos"),
-            followers=res_data.get("followers"),
-            following=res_data.get("following"),
-            created_at=res_data.get("created_at"),
-            avatar_url=res_data.get("avatar_url"),
-            html_url=res_data.get("html_url"),
-            raw=res_data,
-        )
-    except requests.exceptions.RequestException as e:
-        print(f"{RED}Error fetching GitHub user information: {e}{WHITE}")
-        return None
-    except json.JSONDecodeError:
-        print(f"{RED}Error decoding JSON response from GitHub.{WHITE}")
+    provider = GithubProvider()
+    results = provider.execute(github_user)
+
+    if not results:
+        print(f"{RED}No results for '{github_user}'.{WHITE}")
+        print(WHITE + LINES_SEPARATOR)
+        getpass(SPACE_PREFIX + "press enter for back to previous menu ")
         return None
 
+    profile = results[0]
+
+    table_data = [[str(key), str(value)] for key, value in profile.raw.items()]
+    for line_item in tabulate(
+        table_data, headers=["info", "content"], tablefmt="fancy_grid"
+    ).splitlines():
+        print(" " * int(len(SPACE_PREFIX) / 2) + line_item)
+
     print(WHITE + LINES_SEPARATOR)
     getpass(SPACE_PREFIX + "press enter for back to previous menu ")
     return profile