bellingcat/octosuite
1
0
Fork 0
mirror of https://github.com/bellingcat/octosuite.git synced 2026-06-12 05:18:33 +03:00
Code Issues Packages Projects Releases Wiki Activity

Update octosuite.py

Browse Source
This commit is contained in:
Richard Mwewa
2022-11-25 02:52:22 +02:00
committed by GitHub
parent d27d9b9168
commit 29eb2bde95
1 changed files with 98 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

196
octosuite/octosuite.py
View File

@@ -27,59 +27,59 @@ class Octosuite:
# A list of tuples mapping commands to their methods
self.command_map = [("exit", self.exit_session),
("clear", self.clear_screen),
("about", self.about),
("author", self.author),
("help", Help.help_command),
("help:source", Help.source_command),
("help:search", Help.search_command),
("help:user", Help.user_command),
("help:repo", Help.repo_command),
("help:logs", Help.logs_command),
("help:csv", Help.csv_command),
("help:org", Help.org_command),
("source", Help.cource),
("source:tarball", self.download_tarball),
("source:zipball", self.download_zipball),
("org", Help.org),
("org:events", self.org_events),
("org:profile", self.org_profile),
("org:repos", self.org_repos),
("org:member", self.org_member),
("repo", Help.repo),
("repo:path_contents", self.path_contents),
("repo:profile", self.repo_profile),
("repo:contributors", self.repo_contributors),
("repo:stargazers", self.repo_stargazers),
("repo:forks", self.repo_forks),
("repo:issues", self.repo_issues),
("repo:releases", self.repo_releases),
("user", Help.user),
("user:repos", self.user_repos),
("user:gists", self.user_gists),
("user:orgs", self.user_orgs),
("user:profile", self.user_profile),
("user:events", self.user_events),
("user:followers", self.user_followers),
("user:follows", self.user_follows),
("user:following", self.user_following),
("user:subscriptions", self.user_subscriptions),
("search", Help.search),
("search:users", self.user_search),
("search:repos", self.repo_search),
("search:topics", self.topic_search),
("search:issues", self.issue_search),
("search:commits", self.commits_search),
("logs", Help.logs),
("logs:view", self.view_logs),
("logs:read", self.read_log),
("logs:delete", self.delete_log),
("logs:clear", self.clear_logs),
("csv", Help.csv),
("csv:view", self.view_csv),
("csv:read", self.read_csv),
("csv:delete", self.delete_csv),
("csv:clear", self.clear_csv)]
("clear", self.clear_screen),
("about", self.about),
("author", self.author),
("help", Help.help_command),
("help:source", Help.source_command),
("help:search", Help.search_command),
("help:user", Help.user_command),
("help:repo", Help.repo_command),
("help:logs", Help.logs_command),
("help:csv", Help.csv_command),
("help:org", Help.org_command),
("source", Help.cource),
("source:tarball", self.download_tarball),
("source:zipball", self.download_zipball),
("org", Help.org),
("org:events", self.org_events),
("org:profile", self.org_profile),
("org:repos", self.org_repos),
("org:member", self.org_member),
("repo", Help.repo),
("repo:path_contents", self.path_contents),
("repo:profile", self.repo_profile),
("repo:contributors", self.repo_contributors),
("repo:stargazers", self.repo_stargazers),
("repo:forks", self.repo_forks),
("repo:issues", self.repo_issues),
("repo:releases", self.repo_releases),
("user", Help.user),
("user:repos", self.user_repos),
("user:gists", self.user_gists),
("user:orgs", self.user_orgs),
("user:profile", self.user_profile),
("user:events", self.user_events),
("user:followers", self.user_followers),
("user:follows", self.user_follows),
("user:following", self.user_following),
("user:subscriptions", self.user_subscriptions),
("search", Help.search),
("search:users", self.users_search),
("search:repos", self.repos_search),
("search:topics", self.topics_search),
("search:issues", self.issues_search),
("search:commits", self.commits_search),
("logs", Help.logs),
("logs:view", self.view_logs),
("logs:read", self.read_log),
("logs:delete", self.delete_log),
("logs:clear", self.clear_logs),
("csv", Help.csv),
("csv:view", self.view_csv),
("csv:read", self.read_csv),
("csv:delete", self.delete_csv),
("csv:clear", self.clear_csv)]
# Path attribute
self.path_attrs = ['size', 'type', 'path', 'sha', 'html_url']
@@ -325,7 +325,7 @@ class Octosuite:
"""
pass
else:
xprint(f"{MessagePrefix.info} A new release of Octosuite is available ({response['tag_name']}). Run 'pip install --upgrade octosuite' to get the updates.\n")
xprint(f"[{green}UPDATE{reset}] A new release of Octosuite is available ({response['tag_name']}). Run 'pip install --upgrade octosuite' to get the updates.\n")
"""
@@ -361,7 +361,7 @@ class Octosuite:
# Fetching organization info
def org_profile(self):
xprint(f"{white}>> @{green}Organization {white}(username){reset} ", end="")
xprint(f"{white}@{green}Organization {white}(username):{reset} ", end="")
organization = input()
response = requests.get(f"{self.endpoint}/orgs/{organization}")
if response.status_code == 404:
@@ -378,7 +378,7 @@ class Octosuite:
# Fetching user information
def user_profile(self):
xprint(f"{white}>> @{green}Username{reset} ", end="")
xprint(f"{white}@{green}username:{reset} ", end="")
username = input()
response = requests.get(f"{self.endpoint}/users/{username}")
if response.status_code == 404:
@@ -485,7 +485,7 @@ class Octosuite:
def repo_forks(self):
xprint(f"{white}>> %{green}Repository{reset} ", end="")
repo_name = input()
xprint(f"{white}>> @{green}Owner{white} (username){reset} ", end="")
xprint(f"{white}@{green}Owner{white} (username):{reset} ", end="")
username = input()
xprint(MessagePrefix.prompt, LogRoller.limit_output.format("repository forks"), end="")
limit = int(input())
@@ -556,7 +556,7 @@ class Octosuite:
# Fetching organization repositories
def org_repos(self):
xprint(f"{white}>> @{green}Organization{white} (username){reset} ", end="")
xprint(f"{white}@{green}organization{white} (username):{reset} ", end="")
organization = input()
xprint(MessagePrefix.prompt, LogRoller.limit_output.format("organization repositories"), end="")
limit = int(input())
@@ -576,7 +576,7 @@ class Octosuite:
# organization events
def org_events(self):
xprint(f"{white}>> @{green}Organization{white} (username){reset} ", end="")
xprint(f"{white}@{green}organization{white} (username):{reset} ", end="")
organization = input()
xprint(MessagePrefix.prompt, LogRoller.limit_output.format("organization repositories"), end="")
limit = int(input())
@@ -597,9 +597,9 @@ class Octosuite:
# organization member
def org_member(self):
xprint(f"{white}>> @{green}Organization{white} (username){reset} ", end="")
xprint(f"{white}@{green}organization{white} (username):{reset} ", end="")
organization = input()
xprint(f"{white}>> @{green}Username{reset} ", end="")
xprint(f"{white}@{green}username:{reset} ", end="")
username = input()
response = requests.get(f"{self.endpoint}/orgs/{organization}/public_members/{username}")
if response.status_code == 204:
@@ -610,7 +610,7 @@ class Octosuite:
# Fetching user repositories
def user_repos(self):
xprint(f"{white}>> @{green}Username{reset} ", end="")
xprint(f"{white}@{green}username:{reset} ", end="")
username = input()
xprint(MessagePrefix.prompt, LogRoller.limit_output.format("repositories"), end="")
limit = int(input())
@@ -630,7 +630,7 @@ class Octosuite:
# Fetching user's gists
def user_gists(self):
xprint(f"{white}>> @{green}Username{reset} ", end="")
xprint(f"{white}@{green}username:{reset} ", end="")
username = input()
xprint(MessagePrefix.prompt, LogRoller.limit_output.format('gists'), end="")
limit = int(input())
@@ -652,7 +652,7 @@ class Octosuite:
# Fetching a list of organizations that a user owns or belongs to
def user_orgs(self):
xprint(f"{white}>> @{green}Username{reset} ", end="")
xprint(f"{white}@{green}username:{reset} ", end="")
username = input()
xprint(MessagePrefix.prompt, LogRoller.limit_output.format("user organizations"), end="")
limit = int(input())
@@ -674,7 +674,7 @@ class Octosuite:
# Fetching a users events
def user_events(self):
xprint(f"{white}>> @{green}Username{reset} ", end="")
xprint(f"{white}@{green}username:{reset} ", end="")
username = input()
xprint(MessagePrefix.prompt, LogRoller.limit_output.format("events"), end="")
limit = int(input())
@@ -697,7 +697,7 @@ class Octosuite:
# Fetching a target user's subscriptions
def user_subscriptions(self):
xprint(f"{white}>> @{green}Username{reset} ", end="")
xprint(f"{white}@{green}username:{reset} ", end="")
username = input().lower()
xprint(MessagePrefix.prompt, LogRoller.limit_output.format("user subscriptions"), end="")
limit = int(input())
@@ -719,7 +719,7 @@ class Octosuite:
# Fetching a list of users the target follows
def user_following(self):
xprint(f"{white}>> @{green}Username{reset} ", end="")
xprint(f"{white}@{green}username:{reset} ", end="")
username = input().lower()
xprint(MessagePrefix.prompt, LogRoller.limit_output.format("user' following"), end="")
limit = int(input())
@@ -741,7 +741,7 @@ class Octosuite:
# Fetching user's followers
def user_followers(self):
xprint(f"{white}>> @{green}Username{reset} ", end="")
xprint(f"{white}@{green}username:{reset} ", end="")
username = input().lower()
xprint(MessagePrefix.prompt, LogRoller.limit_output.format("user followers"), end="")
limit = int(input())
@@ -763,9 +763,9 @@ class Octosuite:
# Checking whether user[A] follows user[B]
def user_follows(self):
xprint(f"{white}>> @{green}user{white}(A) (username){reset} ", end="")
xprint(f"{white}@{green}user{white}(A) (username):{reset} ", end="")
user_a = input()
xprint(f"{white}>> @{green}user{white}(B) (username){reset} ", end="")
xprint(f"{white}@{green}user{white}(B) (username):{reset} ", end="")
user_b = input()
response = requests.get(f"{self.endpoint}/users/{user_a}/following/{user_b}")
if response.status_code == 204:
@@ -776,68 +776,68 @@ class Octosuite:
# User search
def users_search(self):
xprint(f"{white}>> @{green}Query{white} (eg. john){reset} ", end="")
xprint(f"{white}@{green}query{white} (eg. john):{reset} ", end="")
query = input()
xprint(MessagePrefix.prompt, LogRoller.limit_output.format("user search"), end="")
limit = int(input())
response = requests.get(f"{self.endpoint}/search/users?q={query}&per_page={limit}").json()
for user in response['items']:
user_search_tree = Tree("\n" + user['login'])
users_search_tree = Tree("\n" + user['login'])
for attr in self.user_attrs:
user_search_tree.add(f"{self.user_attr_dict[attr]}: {user[attr]}")
xprint(user_search_tree)
users_search_tree.add(f"{self.user_attr_dict[attr]}: {user[attr]}")
xprint(users_search_tree)
CsvLogger.log_users_search(user, query)
# Repository search
def repos_search(self):
xprint(f"{white}>> %{green}Query{white} (eg. git){reset} ", end="")
xprint(f"{white}%{green}query{white} (eg. git):{reset} ", end="")
query = input()
xprint(MessagePrefix.prompt, LogRoller.limit_output.format("repositor[y][ies] search"), end="")
limit = int(input())
response = requests.get(f"{self.endpoint}/search/repositories?q={query}&per_page={limit}").json()
for repository in response['items']:
repo_search_tree = Tree("\n" + repository['full_name'])
repos_search_tree = Tree("\n" + repository['full_name'])
for attr in self.repo_attrs:
repo_search_tree.add(f"{self.repo_attr_dict[attr]}: {repository[attr]}")
xprint(repo_search_tree)
repos_search_tree.add(f"{self.repo_attr_dict[attr]}: {repository[attr]}")
xprint(repos_search_tree)
CsvLogger.log_repos_search(repository, query)
# Topics search
def topics_search(self):
xprint(f"{white}>> #{green}Query{white} (eg. osint){reset} ", end="")
xprint(f"{white}#{green}query{white} (eg. osint):{reset} ", end="")
query = input()
xprint(MessagePrefix.prompt, LogRoller.limit_output.format("topic(s) search"), end="")
limit = int(input())
response = requests.get(f"{self.endpoint}/search/topics?q={query}&per_page={limit}").json()
for topic in response['items']:
topic_search_tree = Tree("\n" + topic['name'])
topics_search_tree = Tree("\n" + topic['name'])
for attr in self.topic_attrs:
topic_search_tree.add(f"{self.topic_attr_dict[attr]}: {topic[attr]}")
xprint(topic_search_tree)
topics_search_tree.add(f"{self.topic_attr_dict[attr]}: {topic[attr]}")
xprint(topics_search_tree)
CsvLogger.log_topics_search(topic, query)
# Issue search
def issues_search(self):
xprint(f"{white}>> !{green}Query{white} (eg. error){reset} ", end="")
xprint(f"{white}!{green}Query{white} (eg. error):{reset} ", end="")
query = input()
xprint(MessagePrefix.prompt, LogRoller.limit_output.format("issue(s) search"), end="")
limit = int(input())
response = requests.get(f"{self.endpoint}/search/issues?q={query}&per_page={limit}").json()
for issue in response['items']:
issue_search_tree = Tree("\n" + issue['title'])
issues_search_tree = Tree("\n" + issue['title'])
for attr in self.repo_issues_attrs:
issue_search_tree.add(f"{self.repo_issues_attr_dict[attr]}: {issue[attr]}")
xprint(issue_search_tree)
issues_search_tree.add(f"{self.repo_issues_attr_dict[attr]}: {issue[attr]}")
xprint(issues_search_tree)
xprint(issue['body'])
CsvLogger.log_issues_search(issue, query)
# Commits search
def commits_search(self):
xprint(f"{white}>> :{green}Query{white} (eg. filename:index.php){reset} ", end="")
xprint(f"{white}:{green}Query{white} (eg. filename:index.php):{reset} ", end="")
query = input()
xprint(MessagePrefix.prompt, LogRoller.limit_output.format("commit(s) search"), end="")
limit = int(input())
@@ -869,7 +869,7 @@ class Octosuite:
# Read a specified csv file
def read_csv(self):
xprint(f"{white}>> {green}.csv {reset}(filename) ", end="")
xprint(f"{green}csv {white}(filename):{reset} ", end="")
csv_file = input()
with open(os.path.join("output", csv_file + ".csv"), "r") as file:
logging.info(LogRoller.reading_csv.format(csv_file))
@@ -879,7 +879,7 @@ class Octosuite:
# Delete a specified csv file
def delete_csv(self):
xprint(f"{white}>> {green}.csv {reset}filename{reset} ", end="")
xprint(f"{green}csv {white}(filename):{reset} ", end="")
csv_file = input()
os.remove(os.path.join("output", csv_file))
logging.info(LogRoller.deleted_csv.format(csv_file))
@@ -911,7 +911,7 @@ class Octosuite:
# Read a specified log file
def read_log(self):
xprint(f"{white}>> {green}.log date{reset} (eg. 2022-04-27 10:09:36AM) ", end="")
xprint(f"{green}log date{white} (eg. 2022-04-27 10:09:36AM):{reset} ", end="")
log_file = input()
with open(os.path.join(".logs", log_file + ".log"), "r") as log:
logging.info(LogRoller.reading_log.format(log_file))
@@ -920,7 +920,7 @@ class Octosuite:
# Delete a specified log file
def delete_log(self):
xprint(f"{white}>> {green}.log date{reset} (eg. 2022-04-27 10:09:36AM) ", end="")
xprint(f"{green}log date{white} (eg. 2022-04-27 10:09:36AM):{reset} ", end="")
log_file = input()
os.remove(os.path.join(".logs", log_file))
logging.info(LogRoller.deleted_log.format(log_file))
@@ -976,22 +976,22 @@ class Octosuite:
# About program
def about(self):
about_text = Text(f"""
OCTOSUITE © 2022 Richard Mwewa
about_text = f"""
OCTOSUITE © 2023 Richard Mwewa
An advanced and lightning fast framework for gathering open-source intelligence on GitHub users and organizations.
With over 20+ features, Octosuite only runs on 2 external dependencies, and returns the gathered intelligence in a highly readable format.
Whats new in v{version_tag}?
[fixed] Minor fixes
[improved] Removed width from tables, so that they can auto adjust
[added] Added the 'log:clear' command, which will be used to clear all logs
[added] Added the 'csv:clear' command, which will be used to clear all csv files
[{green}fixed{reset}] Minor fixes
[{green}improved{reset}] Removed width from tables, so that they can auto adjust
[{green}added{reset}] Added the 'log:clear' command, which will be used to clear all logs
[{green}added{reset}] Added the 'csv:clear' command, which will be used to clear all csv files
Read the wiki: https://github.com/bellingcat/octosuite/wiki
GitHub REST API documentation: https://docs.github.com/rest
""")
"""
xprint(about_text)