Option #1: With Compute OS Login privileges

A Google Group has been created and granted Compute OS Login privileges to the pentest-mcp Google Cloud Project being used to run the workshop examples. Visit the group's URL at https://groups.google.com/g/nsfsfs26 and join it with the Google account you will use to access the lab exercise.

Visit the Compute Engine console for the project at: https://console.cloud.google.com/compute/instances?project=pentest-mcp and ssh into one of the Kali VMs that has been brought up on Compute Engine in the pentest-mcp project.

In your home directory, clone the repository the exercises reside in.

git clone https://github.com/wu4f/cs475-src

Option #2: Without Compute OS Login privileges

If you prefer not to add yourself to the Google Group, you may instead use an ssh client to log into one of the Kali VMs that has been brought up on Compute Engine in the pentest-mcp project. The usernames are user{01..10} with passwords ptmcp{01..10}.

Within the repository on the course VM, change into the exercise directory, then create a virtual environment and install the required packages.

cd cs475-src/04_MCP/01_sqlite
uv init --bare
uv add -r requirements.txt

There are two main ways of running an MCP server. One way is to run the MCP server locally and communicate with it over standard input/output (STDIO) while another is to run the MCP server remotely and communicate with it over HTTP.

MCP server

The code below implements the SQLite server. It utilizes the FastMCP package to create the server and instantiates one tool within it called query(). The tool handles queries to a local SQLite database by taking a query string and executing it against a specified database, returning the results. By taking a raw query string, the tool is vulnerable to SQL injection attacks. Note that the description of the tool is provided within the comments of the tool declaration. This description is utilized by the server to instruct clients on how to access the tool. An LLM agent is better equipped to call MCP tools if these descriptions are detailed, specific, and accurate.

vulnerable_sqlite_mcp_server.py

from fastmcp import FastMCP
import sqlite3
import sys

mcp = FastMCP("sqlite")
con = sqlite3.connect('db_data/metactf_users.db')

@mcp.tool()
async def query(query: str) -> list:
    """Query a specified Sqlite3 database. Takes a query string as an input parameter and returns the result of the query."""
    cur = con.cursor()
    res = cur.execute(query)
    con.commit()
    return res.fetchall()

if __name__ == "__main__":
    if sys.argv[1] == 'stdio':
        mcp.run(transport="stdio")
    else:
        mcp.run(transport="http", host="0.0.0.0", port=8080)

MCP client

To leverage the tool that the server now supports, we can adapt our prior agent code to be an MCP client, leveraging LangChain's MCP adapter support to invoke the tool on the server. As the code shows, we first define the server we wish to bring up. In this instance, the path in the repository to the prior server code is specified. Then, in the agent loop, we create a connection to the MCP server and load the MCP server's tool into our agent, before querying it. In doing so, the agent will package an MCP call over STDIO via the session's connection and retrieve the results.

from langchain_mcp_adapters.tools import load_mcp_tools
from mcp import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client
import asyncio

database = "db_data/metactf_users.db"

server = StdioServerParameters(
    command="python",
    args=["vulnerable_sqlite_mcp_server_stdio.py","stdio"]
)

prompt = f"You are a Sqlite3 database look up tool. Perform queries on the database at {database} given the user's input.  Utilize the user input verbatim when sending the query to the database and print the query that was sent to the database"

async def run_agent():
    async with stdio_client(server) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            tools = await load_mcp_tools(session)
            agent = create_react_agent(model=llm, tools=tools, prompt=prompt)
            query = "Who are the users in the database?"
            result = await agent.ainvoke({"messages": [("user", query)]})
            print(f"Agent response: {result}")

if __name__ == "__main__":
    result = asyncio.run(run_agent())

Before we use the agent to answer queries about the database, we'll want to establish 'ground-truth' to ensure the agent is working correctly and not hallucinating incorrect results. Within the repository, load the database using the sqlite3 CLI.

cd db_data
sqlite3 metactf_users.db

Using the CLI, query the database to determine "ground-truth" measurements.

The following queries return the column names of the users table, the number of users in the users table, the usernames that start with the letter 'a', and the admin's password hash that includes the number of iterations the hash was generated with.

PRAGMA table_info(users);
SELECT COUNT(*) from users;
SELECT username FROM users WHERE username LIKE 'a%';
SELECT passhash FROM users WHERE username = 'admin';

Once you've obtained ground truth, go back to the program directory and run the program

cd ..
uv run 01_stdio_mcp_client.py

Now, attempt to interact with the database using the agent with natural language queries asking for the same.

For queries that return incorrect results, modify the question or the model to return the correct result.

Finally, try a complex query with step-by-step instructions such as the one below which returns a hashcat command that performs a dictionary search on the admin's password hash.

Modify the query or the model until the agent is able to return a correct answer to the query.

It is quite dangerous to expose an MCP server like this without proper access control. It is also quite dangerous to simply allow agents to have an MCP server accept arbitrary SQL queries from an agent. Using the agent, show whether the server is vulnerable to attack using the queries below.

Exit out of the agent with a blank line. Note, if the agent is able to delete the table, restore it from the command line via:

git checkout db_data/metactf_users.db

The vulnerable SQLite MCP server allowed arbitrary SQL queries to be executed, giving it excessive agency and making it extremely vulnerable to attack. One can improve the security of the MCP server by limiting the functionality of its tools to just the ones necessary as well as utilizing more secure methods for implementing the tool, such as via a parameterized SQL query. A more secure version is shown below:

secure_sqlite_mcp_server.py

@mcp.tool()
async def fetch_users() -> list:
    """Fetch the users in the database.  Takes no arguments and returns a list of users."""
    cur = con.cursor()
    res = cur.execute('SELECT username from USERS')
    return res.fetchall()

@mcp.tool()
async def fetch_users_pass(username: str) -> str:
   """Useful when you want to fetch a password hash for a particular user.  Takes a username as a string argument.  Returns a JSON string"""
   cur = con.cursor()
   res = cur.execute(f"SELECT passhash FROM users WHERE username = ?;",(username,))

As the code shows, the MCP server only allows one to fetch a list of all of the users and to retrieve the password hash for a given user. To leverage this server, modify the MCP client to utilize the secure server.

server = StdioServerParameters(
    command="python",
    args=["secure_sqlite_mcp_server.py","stdio"]
)

Run the program again and repeat the destructive queries to show they no longer work.:

One can also run an MCP server remotely over the network, thus allowing MCP clients to invoke the tools that the server implements over the network using HTTP. In this exercise, an instance of the secure SQLite MCP server has been deployed in a serverless manner in a container running on Google's Cloud Run. The Dockerfile for the container that has been deployed is shown below.

Dockerfile

FROM python:3.10

WORKDIR /app

COPY requirements.txt .

RUN pip install --no-cache-dir -r requirements.txt

RUN mkdir -p db_data

COPY secure_sqlite_mcp_server.py .

COPY ./db_data/metactf_users.db ./db_data/metactf_users.db

EXPOSE 8080

CMD ["python", "secure_sqlite_mcp_server.py", "http"]

MCP server

Visit Cloud Run and click on the container deployment to obtain its URL.

MCP client

We'll now run the MCP client on the course VM and allow it to utilize the MCP server running in Cloud Run. To begin with, on the course VM, set the MCP_URL environment variable to the URL that is returned by Cloud Run.

export MCP_URL="https:// ...a.run.app"

To adapt the MCP client to utilize the remote MCP server, we simply tweak the client to utilize the Streamable HTTP interface to the MCP server's endpoint URL as shown in the snippet below, keeping the rest of the client the same.

async def run_agent():
        async with streamablehttp_client(f"{os.getenv('MCP_URL')}/mcp/") as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()

            tools = await load_mcp_tools(session)
...

Run the agent and interact with the MCP server on Cloud Run.

uv run 02_http_mcp_client.py

Repeat the destructive queries again:

Then, run the initial queries to show the attacks did not impact the server.

Back in the web interface for Cloud Run, click on the deployed service and then navigate to the logs. Find the requests associated with your queries.