Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,19 @@ The results of the audit are written to an SQLite database, which is opened auto
the end of the run. The results are in the table named "audit_result". The table has a column named "has_vulnerability",
with checkmarks in the rows that are most likely to be genuine vulnerabilities.

## Mobile application support

The audit workflow recognizes Android, iOS, and cross-platform mobile applications alongside web applications and libraries.
Mobile entry-point collection records exported Android components, deep links, URL schemes, Universal Links, app extensions,
and WebView JavaScript bridges, including their permissions, export status, and input filters.

The [audit runner](scripts/audit/run_audit.sh) collects this metadata before classification, which applies mobile-specific
guidance to components marked as mobile applications. The existing `--advisory` and `-m` options also apply to mobile runs.
Both mobile entry-point taskflows use the container-backed source-access toolbox and require Docker.

Mobile metadata is stored in the `mobile_entry_point` table. Existing context databases automatically gain the nullable
`application.is_mobile_app` column without deleting existing results.

## Running with docker script

We recommend running taskflows in a sandboxed environment. [GitHub Codespaces](https://github.com/features/codespaces) are convenient, or if you prefer you can use the script [`run_seclab_agent.sh`](scripts/run_seclab_agent.sh) to run a docker container of the `seclab-taskflow-agent` as outlined [here](https://github.com/GitHubSecurityLab/seclab-taskflow-agent/tree/main?tab=readme-ov-file#deploying-from-docker). Note that this script needs to be run from the main directory of the repo, and the `.env` file with the environment variables for the custom MCP servers to store data needs to be in the same directory.
Expand Down
1 change: 1 addition & 0 deletions scripts/audit/run_audit.sh
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@ fi
python -m seclab_taskflow_agent $MODEL_CONFIG_FLAG -t seclab_taskflows.taskflows.audit.fetch_source_code -g repo="$1"
python -m seclab_taskflow_agent $MODEL_CONFIG_FLAG -t seclab_taskflows.taskflows.audit.identify_applications -g repo="$1"
python -m seclab_taskflow_agent $MODEL_CONFIG_FLAG -t seclab_taskflows.taskflows.audit.gather_web_entry_point_info -g repo="$1"
python -m seclab_taskflow_agent $MODEL_CONFIG_FLAG -t seclab_taskflows.taskflows.audit.gather_mobile_entry_point_info -g repo="$1"
python -m seclab_taskflow_agent $MODEL_CONFIG_FLAG -t seclab_taskflows.taskflows.audit.gather_security_entry_point_info -g repo="$1"

if [ "$USE_ADVISORY" = true ]; then
Expand Down
153 changes: 148 additions & 5 deletions src/seclab_taskflows/mcp_servers/repo_context.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,12 +9,12 @@
import os
import yaml
from sqlalchemy.orm import DeclarativeBase, mapped_column, Mapped
from sqlalchemy import create_engine
from sqlalchemy import create_engine, inspect as sa_inspect, text
from sqlalchemy.orm import Session
from pathlib import Path
from seclab_taskflow_agent.path_utils import mcp_data_dir, log_file_name

from .repo_context_models import Application, EntryPoint, UserAction, WebEntryPoint, ApplicationIssue, AuditResult, SecurityEntryPoint, Base
from .repo_context_models import Application, EntryPoint, UserAction, WebEntryPoint, MobileEntryPoint, ApplicationIssue, AuditResult, SecurityEntryPoint, Base
from .repo_context_models import LowSeverityAuditResult, AuditResultEvidence, ExternalDependency
from .utils import process_repo

Expand All @@ -35,6 +35,7 @@ def app_to_dict(result):
"notes": result.notes,
"is_app": bool(result.is_app),
"is_library": bool(result.is_library),
"is_mobile_app": bool(result.is_mobile_app),
}


Expand Down Expand Up @@ -86,6 +87,21 @@ def security_entry_point_to_dict(sep):
"notes": sep.notes,
}

def mobile_entry_point_to_dict(mep):
return {
"id": mep.id,
"entry_point_id": mep.entry_point_id,
"entry_type": mep.entry_type,
"scheme_or_action": mep.scheme_or_action,
"exported": mep.exported,
"permissions": mep.permissions,
"data_filter": mep.data_filter,
"component": mep.component,
"repo": mep.repo.lower(),
"notes": mep.notes,
}


def audit_result_to_dict(res):
return {
"id": res.id,
Expand Down Expand Up @@ -115,6 +131,7 @@ def __init__(self, memcache_state_dir: str):
EntryPoint.__table__,
UserAction.__table__,
WebEntryPoint.__table__,
MobileEntryPoint.__table__,
ApplicationIssue.__table__,
AuditResult.__table__,
LowSeverityAuditResult.__table__,
Expand All @@ -123,19 +140,38 @@ def __init__(self, memcache_state_dir: str):
ExternalDependency.__table__,
],
)
self._migrate_schema()

def store_new_application(self, repo, location, is_app, is_library, notes):
def _migrate_schema(self):
"""Add any missing columns to existing tables (lightweight migration for SQLite)."""
inspector = sa_inspect(self.engine)
migrations = [
("application", "is_mobile_app", "BOOLEAN"),
]
for table_name, column_name, column_type in migrations:
if not inspector.has_table(table_name):
continue
existing_columns = {col["name"] for col in inspector.get_columns(table_name)}
if column_name not in existing_columns:
logging.info(f"Migrating: adding column {column_name} to {table_name}")
with self.engine.begin() as conn:
conn.execute(text(f"ALTER TABLE {table_name} ADD COLUMN {column_name} {column_type}"))

def store_new_application(self, repo, location, is_app, is_library, notes, is_mobile_app=None):
with Session(self.engine) as session:
existing = session.query(Application).filter_by(repo=repo, location=location).first()
if existing:
if is_app is not None:
existing.is_app = is_app
if is_library is not None:
existing.is_library = is_library
if is_mobile_app is not None:
existing.is_mobile_app = is_mobile_app
existing.notes += notes
else:
new_application = Application(
repo=repo, location=location, is_app=is_app, is_library=is_library, notes=notes
repo=repo, location=location, is_app=is_app, is_library=is_library,
is_mobile_app=is_mobile_app, notes=notes
)
session.add(new_application)
session.commit()
Expand Down Expand Up @@ -267,6 +303,53 @@ def store_new_security_entry_point(self, repo, entry_point_id, type, component,
session.commit()
return f"Updated or added security entry point for entry_point_id {entry_point_id} in {repo}."

def store_new_mobile_entry_point(
self, repo, entry_point_id, entry_type, scheme_or_action, exported, permissions, data_filter, component, notes, update=False
):
with Session(self.engine) as session:
existing = session.query(MobileEntryPoint).filter_by(repo=repo, entry_point_id=entry_point_id).first()
if existing:
existing.notes += notes
if entry_type:
existing.entry_type = entry_type
if scheme_or_action:
existing.scheme_or_action = scheme_or_action
if exported is not None:
existing.exported = exported
if permissions:
existing.permissions = permissions
if data_filter:
existing.data_filter = data_filter
if component is not None:
existing.component = component
else:
if update:
return f"No mobile entry point exists at repo {repo} with entry_point_id {entry_point_id}."
new_mobile_entry_point = MobileEntryPoint(
repo=repo,
entry_point_id=entry_point_id,
entry_type=entry_type,
scheme_or_action=scheme_or_action,
exported=exported,
permissions=permissions,
data_filter=data_filter,
component=component,
notes=notes,
)
session.add(new_mobile_entry_point)
session.commit()
return f"Updated or added mobile entry point for entry_point_id {entry_point_id} in {repo}."

def get_mobile_entries(self, repo, component_id):
with Session(self.engine) as session:
results = session.query(MobileEntryPoint).filter_by(repo=repo, component=component_id).all()
return [mobile_entry_point_to_dict(r) for r in results]

def get_mobile_entries_for_repo(self, repo):
with Session(self.engine) as session:
results = session.query(MobileEntryPoint).filter_by(repo=repo).all()
return [mobile_entry_point_to_dict(r) for r in results]

def store_new_user_action(self, repo, app_id, file, line, notes, update=False):
with Session(self.engine) as session:
existing = session.query(UserAction).filter_by(repo=repo, file=file, line=line).first()
Expand Down Expand Up @@ -517,6 +600,7 @@ def clear_repo(self, repo):
session.query(UserAction).filter_by(repo=repo).delete()
session.query(ApplicationIssue).filter_by(repo=repo).delete()
session.query(WebEntryPoint).filter_by(repo=repo).delete()
session.query(MobileEntryPoint).filter_by(repo=repo).delete()
session.query(AuditResult).filter_by(repo=repo).delete()
session.query(LowSeverityAuditResult).filter_by(repo=repo).delete()
session.query(SecurityEntryPoint).filter_by(repo=repo).delete()
Expand Down Expand Up @@ -544,11 +628,14 @@ def store_new_component(
is_app: bool = Field(description="Is this an application", default=None),
is_library: bool = Field(description="Is this a library", default=None),
notes: str = Field(description="The notes taken for this component", default=""),
is_mobile_app: bool = Field(description="Is this a mobile application (e.g. Android, iOS)", default=None),
):
"""
Stores a new component in the database.
"""
return backend.store_new_application(process_repo(owner, repo), location, is_app, is_library, notes)
return backend.store_new_application(
process_repo(owner, repo), location, is_app, is_library, notes, is_mobile_app=is_mobile_app
)


@mcp.tool()
Expand Down Expand Up @@ -849,6 +936,62 @@ def get_security_entry_points_for_repo(
repo = process_repo(owner, repo)
return json.dumps(backend.get_security_entries_for_repo(repo))

@mcp.tool()
def store_new_mobile_entry_point(
owner: str = Field(description="The owner of the GitHub repository"),
repo: str = Field(description="The name of the GitHub repository"),
entry_point_id: int = Field(description="The ID of the entry point this mobile entry point refers to"),
location: str = Field(description="The directory of the component; required when component is not supplied", default=""),
entry_type: str = Field(description="Type of mobile entry point: deep_link, intent, url_scheme, content_provider, broadcast_receiver, js_bridge, app_extension, universal_link, etc.", default=""),
scheme_or_action: str = Field(description="URL scheme (e.g. myapp://) or intent action (e.g. android.intent.action.VIEW)", default=""),
exported: bool = Field(description="Whether the Android component is exported (accessible to other apps)", default=None),
permissions: str = Field(description="Required permissions to access this entry point, if any", default=""),
data_filter: str = Field(description="Intent data filter, URL path pattern, host, etc.", default=""),
component: int = Field(description="Component identifier", default=0),
notes: str = Field(description="Notes for this mobile entry point", default=""),
):
"""
Stores a new mobile entry point in a component to the database. A mobile entry point extends a regular entry point
with mobile-specific properties like entry type, URL scheme/intent action, export status, permissions, and data filters.
"""
repo = process_repo(owner, repo)
if component == 0:
if not location:
return "Error: Provide component or location for this mobile entry point."
app = backend.get_app(repo, location)
if not app:
return f"Error: No component exists in repo: {repo} and location {location}"
component = app.id
return backend.store_new_mobile_entry_point(
repo, entry_point_id, entry_type, scheme_or_action, exported, permissions, data_filter, component, notes
)


@mcp.tool()
def get_mobile_entry_points_component(
owner: str = Field(description="The owner of the GitHub repository"),
repo: str = Field(description="The name of the GitHub repository"),
component_id: int = Field(description="The ID of the component"),
):
"""
Get all mobile entry points for a component
"""
repo = process_repo(owner, repo)
return json.dumps(backend.get_mobile_entries(repo, component_id))


@mcp.tool()
def get_mobile_entry_points_for_repo(
owner: str = Field(description="The owner of the GitHub repository"),
repo: str = Field(description="The name of the GitHub repository"),
):
"""
Get all mobile entry points of a repo
"""
repo = process_repo(owner, repo)
return json.dumps(backend.get_mobile_entries_for_repo(repo))


@mcp.tool()
def get_user_actions(
owner: str = Field(description="The owner of the GitHub repository"),
Expand Down
30 changes: 28 additions & 2 deletions src/seclab_taskflows/mcp_servers/repo_context_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,12 +19,13 @@ class Application(Base):
notes: Mapped[str] = mapped_column(Text)
is_app: Mapped[bool] = mapped_column(nullable=True)
is_library: Mapped[bool] = mapped_column(nullable=True)
is_mobile_app: Mapped[bool] = mapped_column(nullable=True)

def __repr__(self):
return (
f"<Application(id={self.id}, repo={self.repo}, "
f"location={self.location}, is_app={self.is_app}, is_library={self.is_library}"
f"notes={self.notes}>"
f"location={self.location}, is_app={self.is_app}, is_library={self.is_library}, "
f"is_mobile_app={self.is_mobile_app}, notes={self.notes}>"
)


Expand Down Expand Up @@ -148,6 +149,31 @@ def __repr__(self):
f"untrusted_input={self.untrusted_input}, notes={self.notes}, repo={self.repo})>"
)

class MobileEntryPoint(Base):
"""An entry point specific to mobile applications (e.g. deep links, intents, URL schemes)."""
__tablename__ = "mobile_entry_point"

id: Mapped[int] = mapped_column(primary_key=True)
entry_point_id = Column(Integer, ForeignKey("entry_point.id", ondelete="CASCADE"))
entry_type: Mapped[str] # deep_link, intent, url_scheme, content_provider, broadcast_receiver, js_bridge, etc.
scheme_or_action: Mapped[str] # URL scheme (myapp://) or intent action (android.intent.action.VIEW)
exported: Mapped[bool] = mapped_column(nullable=True) # whether the component is exported (Android)
permissions: Mapped[str] # required permissions, if any
data_filter: Mapped[str] # intent data filter, URL path pattern, etc.
component: Mapped[int] # component id
notes: Mapped[str] = mapped_column(Text)
repo: Mapped[str]

def __repr__(self):
return (
f"<MobileEntryPoint(entry_point_id={self.entry_point_id}, "
f"entry_type={self.entry_type}, scheme_or_action={self.scheme_or_action}, "
f"exported={self.exported}, permissions={self.permissions}, "
f"data_filter={self.data_filter}, component={self.component}, "
f"notes={self.notes}, repo={self.repo})>"
)


class UserAction(Base):
__tablename__ = "user_action"

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,15 @@ taskflow:
An authentication/authorization component may be susceptible to IDOR, bypasses etc.
Components that grant user access to data may require access control or authentication.

For mobile applications, consider risks such as:
- Deep link / URL scheme hijacking or mishandling that triggers sensitive actions without validation
- Exported Android components (Activities, Services, BroadcastReceivers, ContentProviders) accessible
to malicious apps via crafted intents
- WebViews loading untrusted content with JavaScript bridges that expose native functionality
- Insecure data storage (tokens, credentials, PII in world-readable or unencrypted locations)
- Content Provider exposure allowing other apps to read or modify sensitive data
- Path traversal via file handling from untrusted sources (downloads, shared content, zip archives)

We're looking for more concrete and serious security issues that affects system integrity or
lead to information leak, so please do not include issues like brute force, Dos, log injection etc.

Expand Down
Loading
Loading