Usage Guide

This guide provides practical examples of how to use the ctfbridge library for common Capture The Flag (CTF) tasks. Each section focuses on a specific aspect of the library.

Asynchronous Code

All examples in this guide use async and await because CTFBridge is designed to be asynchronous. Ensure you run these within an async function and use asyncio.run() or an existing event loop.

Table of Contents

• Initializing the Client
• Authentication
• Working with Challenges
• Handling Attachments
• Accessing the Scoreboard
• Error Handling

---

Initializing the Client 🚀

The create_client function is your entry point.

Automatic Platform Detection

This is the simplest way. CTFBridge inspects the URL to identify the platform.

import asyncio

from ctfbridge import create_client


async def main():
    # Auto-detects the platform (e.g., CTFd, rCTF) from the URL
    client = await create_client("https://demo.ctfd.io")
    print(
        f"Successfully created client for: {client.platform_url} (Platform: {client.platform_name})"
    )

    # You can now use the client to interact with the platform
    # e.g., await client.auth.login(...)
    #       challenges = await client.challenges.get_all()


if __name__ == "__main__":
    asyncio.run(main())

Specifying a Platform

If auto-detection fails or you want to be explicit:

import asyncio

from ctfbridge import create_client
from ctfbridge.exceptions import UnknownBaseURLError, UnknownPlatformError


async def main():
    # Explicitly specify the platform and URL
    # This is useful if to reduce time spent identifying the platform.
    try:
        client = await create_client("https://demo.ctfd.io", platform="ctfd")
        print(
            f"Successfully created client for: {client.platform_url} (Platform: {client.platform_name})"
        )

        # Example of a potentially incorrect platform specification
        # client_rctf = await create_client("https://demo.ctfd.io", platform="rctf")
        # print(f"Client: {client_rctf.platform_url} (Platform: {client_rctf.platform_name})")

    except UnknownPlatformError as e:
        print(f"Error: {e}")
    except UnknownBaseURLError as e:
        print(f"Error: Could not determine base URL for {e.url}")
    except Exception as e:
        print(f"An unexpected error occurred: {e}")


if __name__ == "__main__":
    asyncio.run(main())

---

Authentication 🔑

Accessing challenges or submitting flags often requires logging in.

Login with Credentials

Primarily for platforms like CTFd.

import asyncio

from ctfbridge import create_client
from ctfbridge.exceptions import CTFBridgeError, LoginError


async def main():
    # Initialize client (CTFd in this example)
    client = await create_client("https://demo.ctfd.io")

    try:
        # Attempt to login with username and password
        await client.auth.login(username="user", password="passworda")
        print("Login successful!")

    except LoginError as e:
        print(f"Login failed: {e}")
    except CTFBridgeError as e:
        print(f"An error occurred: {e}")
    except Exception as e:
        print(f"An unexpected error occurred: {e}")


if __name__ == "__main__":
    asyncio.run(main())

Logging Out

Clears session cookies and authorization headers.

import asyncio

from ctfbridge import create_client
from ctfbridge.exceptions import CTFBridgeError, LoginError


async def main():
    client = await create_client("https://demo.ctfd.io")

    try:
        try:
            await client.auth.login(username="user", password="password")
            print("Login successful (or seemed to be).")
        except LoginError:
            print("Login failed.")

        await client.auth.logout()
        print("Logout successful! Session cookies and auth headers are cleared.")

    except CTFBridgeError as e:
        print(f"A CTFBridge error occurred: {e}")
    except Exception as e:
        print(f"An unexpected error occurred: {e}")


if __name__ == "__main__":
    asyncio.run(main())

---

Working with Challenges 🧩

Fetching All Challenges

import asyncio

from ctfbridge import create_client
from ctfbridge.exceptions import CTFBridgeError, UnauthorizedError


async def main():
    # Initialize client (works with CTFd, rCTF, HTB etc.)
    client = await create_client("https://demo.ctfd.io")
    # Authenticate to the platform
    await client.auth.login(username="user", password="password")  # Or token

    try:
        print("Fetching all challenges (basic details)...")
        # detailed=False fetches only basic info, usually faster.
        # enrich=False skips client-side enrichment
        # (parsing authors, attachments, services from description).
        challenges_basic = await client.challenges.get_all(detailed=False, enrich=False)
        if challenges_basic:
            print(f"Found {len(challenges_basic)} challenges (basic details):")
            for chal in challenges_basic[:3]:  # Print first 3
                print(
                    f"  ID: {chal.id}, Name: {chal.name}, Category: {chal.category}, Points: {chal.value}, Solved: {chal.solved}"
                )
        else:
            print("No challenges found or platform requires authentication.")

        print("\nFetching all challenges (detailed and enriched)...")
        # detailed=True fetches full details (might involve more requests per challenge on some platforms).
        # enrich=True applies client-side parsers to extract more info from descriptions.
        challenges_detailed = await client.challenges.get_all(detailed=True, enrich=True)
        if challenges_detailed:
            print(f"Found {len(challenges_detailed)} challenges (detailed):")
            for chal in challenges_detailed[:3]:  # Print first 3
                print(f"  ID: {chal.id}, Name: {chal.name}")
                print(f"    Category: {chal.category} (Normalized: {chal.normalized_category})")
                print(f"    Points: {chal.value}, Solved: {chal.solved}")
                print(f"    Description: {chal.description[:20]}...")
                print(f"    Authors: {chal.authors}")
                if chal.attachments:
                    print(f"    Attachments: {[att.name for att in chal.attachments]}")
        else:
            print("No challenges found.")

    except UnauthorizedError:
        print("Error: This operation requires authentication. Please login first.")
    except CTFBridgeError as e:
        print(f"A CTFBridge error occurred: {e}")
    except Exception as e:
        print(f"An unexpected error occurred: {e}")


if __name__ == "__main__":
    asyncio.run(main())

Fetching a Challenge by ID

import asyncio

from ctfbridge import create_client
from ctfbridge.exceptions import ChallengeFetchError, CTFBridgeError, UnauthorizedError


async def main():
    client = await create_client("https://demo.ctfd.io")
    await client.auth.login(username="user", password="password")

    # --- Get a known challenge ID from the platform ---
    # First, let's get all challenges to find a valid ID to query
    challenge_id_to_fetch = None
    try:
        print("Fetching a list of challenges to get a valid ID...")
        all_challenges = await client.challenges.get_all(detailed=False)
        if all_challenges:
            challenge_id_to_fetch = all_challenges[0].id  # Get the ID of the first challenge
            print(f"Will attempt to fetch details for challenge ID: {challenge_id_to_fetch}")
        else:
            print(
                "No challenges found to get an ID from. Make sure the platform is accessible and has challenges."
            )
            return
    except CTFBridgeError as e:
        print(f"Error fetching initial challenge list: {e}")
        return
    # --- End of getting a challenge ID ---

    if not challenge_id_to_fetch:
        print("Could not obtain a challenge ID to fetch.")
        return

    try:
        print(f"\nFetching challenge by ID: {challenge_id_to_fetch} (enriched)...")
        # enrich=True (default) applies client-side parsers
        challenge = await client.challenges.get_by_id(challenge_id_to_fetch)

        if challenge:
            print(f"Successfully fetched challenge: {challenge.name}")
            print(f"  ID: {challenge.id}")
            print(f"  Category: {challenge.category} (Normalized: {challenge.normalized_category})")
            print(f"  Points: {challenge.value}")
            print(f"  Solved: {challenge.solved}")
            print(f"  Description: {challenge.description[:200]}...")
            print(f"  Authors: {challenge.authors}")
            if challenge.attachments:
                print("  Attachments:")
                for att in challenge.attachments:
                    print(f"    - Name: {att.name}, URL: {att.url}")
        else:
            print(f"Challenge with ID {challenge_id_to_fetch} not found.")

    except ChallengeFetchError as e:
        print(f"Error fetching challenge by ID: {e}")
    except UnauthorizedError:
        print("Error: This operation requires authentication. Please login first.")
    except CTFBridgeError as e:
        print(f"A CTFBridge error occurred: {e}")
    except Exception as e:
        print(f"An unexpected error occurred: {e}")


if __name__ == "__main__":
    asyncio.run(main())

Filtering Challenges

Apply filters directly in get_all():

import asyncio

from ctfbridge import create_client
from ctfbridge.exceptions import CTFBridgeError


async def main():
    client = await create_client("https://demo.ctfd.io")
    await client.auth.login(username="user", password="password")

    try:
        print("Fetching all challenges to demonstrate filtering...")
        all_challenges = await client.challenges.get_all(detailed=True, enrich=True)
        if not all_challenges:
            print("No challenges found to filter.")
            return
        print(f"Total challenges fetched: {len(all_challenges)}")

        # Example 1: Filter by category "Pwn" (case-insensitive for name_contains, exact for category)
        # Note: demo.ctfd.io might not have a "Pwn" category. Adjust as needed.
        # We'll use a category that likely exists, e.g., the category of the first challenge.
        example_category = all_challenges[0].category if all_challenges else "Miscellaneous"
        print(f"\nFiltering for category: '{example_category}'")
        pwn_challenges = await client.challenges.get_all(category=example_category, detailed=False)
        print(f"Found {len(pwn_challenges)} challenges in category '{example_category}':")
        for chal in pwn_challenges[:3]:
            print(f"  - {chal.name} ({chal.value} pts)")

        # Example 2: Filter by minimum points (e.g., >= 100 points)
        min_pts = 100
        print(f"\nFiltering for challenges with >= {min_pts} points...")
        high_value_challenges = await client.challenges.get_all(min_points=min_pts, detailed=False)
        print(f"Found {len(high_value_challenges)} challenges with at least {min_pts} points:")
        for chal in high_value_challenges[:3]:
            print(f"  - {chal.name} ({chal.value} pts)")

        # Example 3: Filter by solved status (e.g., unsolved challenges)
        # On demo.ctfd.io without login, 'solved' will likely be False for all.
        print("\nFiltering for unsolved challenges...")
        unsolved_challenges = await client.challenges.get_all(solved=False, detailed=False)
        print(f"Found {len(unsolved_challenges)} unsolved challenges:")
        for chal in unsolved_challenges[:3]:
            print(f"  - {chal.name} ({chal.value} pts, Solved: {chal.solved})")

        # Example 4: Filter by name containing "Web" (case-insensitive)
        search_term = "Web"
        print(f"\nFiltering for challenges with name containing '{search_term}'...")
        web_challenges_by_name = await client.challenges.get_all(
            name_contains=search_term, detailed=False
        )
        print(f"Found {len(web_challenges_by_name)} challenges with '{search_term}' in name:")
        for chal in web_challenges_by_name[:3]:
            print(f"  - {chal.name} ({chal.value} pts)")

        # Example 5: Combined filters - category "Web" (if exists) AND points > 50
        # Adjust category if "Web" doesn't exist on demo.ctfd.io
        target_category_for_combo = "Web"  # or another existing category
        # Check if this category exists, otherwise pick one
        if not any(c.category == target_category_for_combo for c in all_challenges):
            target_category_for_combo = (
                all_challenges[0].category if all_challenges else "Miscellaneous"
            )

        print(f"\nFiltering for category '{target_category_for_combo}' AND points > 50...")
        combined_filter_chals = await client.challenges.get_all(
            category=target_category_for_combo, min_points=51, detailed=False
        )
        print(f"Found {len(combined_filter_chals)} challenges matching combined filters:")
        for chal in combined_filter_chals[:3]:
            print(f"  - {chal.name} (Category: {chal.category}, Points: {chal.value})")

    except CTFBridgeError as e:
        print(f"A CTFBridge error occurred: {e}")
    except Exception as e:
        print(f"An unexpected error occurred: {e}")


if __name__ == "__main__":
    asyncio.run(main())

Submitting Flags

Requires authentication and platform support.

import asyncio

from ctfbridge import create_client
from ctfbridge.exceptions import (
    CTFBridgeError,
    CTFInactiveError,
    LoginError,
    RateLimitError,
    SubmissionError,
    UnauthorizedError,
)


async def main():
    client = await create_client("https://demo.ctfd.io")

    username = "user"
    password = "password"

    challenge_id_to_submit = None
    flag_to_submit = "CTF{dummy_flag_for_testing}"  # Replace with a real flag if testing a solve

    try:
        # 1. Login
        print(f"Attempting to login as {username}...")
        await client.auth.login(username=username, password=password)
        print("Login successful!")

        # 2. Get a challenge ID to submit to
        # For a real scenario, you'd solve the challenge and know its ID.
        # Here, we'll just pick the first available challenge.
        print("Fetching challenges to get an ID...")
        challenges = await client.challenges.get_all(detailed=False)
        if not challenges:
            print("No challenges found. Cannot proceed with flag submission.")
            return
        challenge_id_to_submit = challenges[0].id
        challenge_name_to_submit = challenges[0].name
        print(
            f"Will attempt to submit a flag to challenge ID: {challenge_id_to_submit} ('{challenge_name_to_submit}')"
        )

        # 3. Submit the flag
        print(f"Submitting flag '{flag_to_submit}' to challenge ID {challenge_id_to_submit}...")
        result = await client.challenges.submit(
            challenge_id=challenge_id_to_submit, flag=flag_to_submit
        )

        print("\n--- Submission Result ---")
        print(f"Correct: {result.correct}")
        print(f"Message: {result.message}")
        print()

        if result.correct:
            print("Congratulations! Flag was correct.")
        else:
            print("Flag was incorrect or already submitted.")

    except LoginError as e:
        print(f"Login failed: {e}")
    except UnauthorizedError:
        print(
            "Error: Authentication is required for this action, but login might have failed silently or token expired."
        )
    except SubmissionError as e:
        print(f"Flag submission failed: {e.reason}")
        print(f"  Challenge ID: {e.challenge_id}, Flag: {e.flag}")
    except CTFInactiveError as e:
        print(f"CTF Inactive: {e}")
    except RateLimitError as e:
        print(f"Rate Limited: {e}. Retry after: {e.retry_after}s")
    except CTFBridgeError as e:
        print(f"A CTFBridge error occurred: {e}")
    except Exception as e:
        print(f"An unexpected error occurred: {e}")


if __name__ == "__main__":
    asyncio.run(main())

---

Handling Attachments 📂

Download files associated with challenges.

Downloading a Single Attachment

import asyncio
import os
import tempfile

from ctfbridge import create_client
from ctfbridge.exceptions import AttachmentDownloadError, ChallengeFetchError, CTFBridgeError
from ctfbridge.models.challenge import Attachment


async def main():
    client = await create_client("https://demo.ctfd.io")
    await client.auth.login(username="user", password="password")

    attachment_to_download = None
    challenge_name = None

    try:
        print("Fetching challenges to find one with an attachment...")
        challenges = await client.challenges.get_all(
            detailed=True, has_attachments=True, enrich=True
        )
        if not challenges:
            print("No challenges found.")
            return

        for chal in challenges:
            if chal.attachments:
                attachment_to_download = chal.attachments[0]
                challenge_name = chal.name
                print(
                    f"Found challenge '{challenge_name}' with attachment: '{attachment_to_download.name}' ({attachment_to_download.url})"
                )
                break

        if not attachment_to_download:
            print("No challenges with attachments found")
            return

        # Create a temporary directory to save the attachment
        with tempfile.TemporaryDirectory() as tmpdir:
            print(
                f"\nAttempting to download attachment '{attachment_to_download.name}' to {tmpdir}..."
            )

            # Basic download
            try:
                saved_path = await client.attachments.download(
                    attachment_to_download, save_dir=tmpdir
                )
                print(f"Attachment downloaded successfully to: {saved_path}")
                assert os.path.exists(saved_path)
            except AttachmentDownloadError as e:
                print(f"Error downloading attachment: {e}")
            except Exception as e:
                print(f"An unexpected error occurred during basic download: {e}")

            # Download with a custom filename
            custom_filename = f"custom_{attachment_to_download.name}"
            print(
                f"\nAttempting to download attachment with custom name '{custom_filename}' to {tmpdir}..."
            )
            try:
                saved_path_custom = await client.attachments.download(
                    attachment_to_download, save_dir=tmpdir, filename=custom_filename
                )
                print(
                    f"Attachment downloaded successfully with custom name to: {saved_path_custom}"
                )
                assert os.path.exists(saved_path_custom)
                assert os.path.basename(saved_path_custom) == custom_filename
            except AttachmentDownloadError as e:
                print(f"Error downloading attachment with custom name: {e}")
            except Exception as e:
                print(f"An unexpected error occurred during custom filename download: {e}")

    except ChallengeFetchError as e:
        print(f"Error fetching challenges: {e}")
    except CTFBridgeError as e:
        print(f"A CTFBridge error occurred: {e}")
    except Exception as e:
        print(f"An unexpected error occurred: {e}")


if __name__ == "__main__":
    asyncio.run(main())

Downloading All Attachments for a Challenge

import asyncio
import os
import tempfile

from ctfbridge import create_client
from ctfbridge.exceptions import AttachmentDownloadError, ChallengeFetchError, CTFBridgeError


async def main():
    client = await create_client("https://demo.ctfd.io")
    await client.auth.login(username="user", password="password")

    challenge_with_attachments = None

    try:
        print("Fetching challenges to find one with multiple attachments (or any attachments)...")
        challenges = await client.challenges.get_all(detailed=True, enrich=True)
        if not challenges:
            print("No challenges found.")
            return

        for chal in challenges:
            if chal.attachments:  # Could be one or more
                challenge_with_attachments = chal
                print(f"Found challenge '{chal.name}' with {len(chal.attachments)} attachment(s).")
                for att in chal.attachments:
                    print(f"  - Attachment: '{att.name}', URL: '{att.url}'")
                break  # Take the first challenge found with any attachments

        if not challenge_with_attachments:
            print("No challenges with attachments found on the platform.")
            return

        # Create a temporary directory to save the attachments
        # You can use a specific path like "./challenge_downloads" instead of tempfile
        with tempfile.TemporaryDirectory() as tmpdir:
            challenge_save_dir = os.path.join(
                tmpdir, challenge_with_attachments.name.replace(" ", "_")
            )  # Sanitize name for dir
            print(
                f"\nAttempting to download all attachments for '{challenge_with_attachments.name}' to {challenge_save_dir}..."
            )

            try:
                # The download_all method takes a list of Attachment objects
                downloaded_paths = await client.attachments.download_all(
                    attachments=challenge_with_attachments.attachments, save_dir=challenge_save_dir
                )

                if downloaded_paths:
                    print("Successfully downloaded the following files:")
                    for path in downloaded_paths:
                        print(f"  - {path}")
                        assert os.path.exists(path)
                else:
                    print(
                        "No files were downloaded. This might happen if all downloads failed or there were no valid attachments."
                    )

            except AttachmentDownloadError as e:
                # This might be raised if a global issue occurs, though individual errors are often logged and skipped by download_all
                print(f"A general error occurred during batch download: {e}")
            except Exception as e:
                print(f"An unexpected error occurred during download_all: {e}")

    except ChallengeFetchError as e:
        print(f"Error fetching challenges: {e}")
    except CTFBridgeError as e:
        print(f"A CTFBridge error occurred: {e}")
    except Exception as e:
        print(f"An unexpected error occurred: {e}")


if __name__ == "__main__":
    asyncio.run(main())

---

Accessing the Scoreboard 🏆

View top teams or users.

import asyncio

from ctfbridge import create_client
from ctfbridge.exceptions import (
    CTFBridgeError,
    CTFInactiveError,
    ScoreboardFetchError,
    UnauthorizedError,
)


async def main():
    # Using demo.ctfd.io, which usually has a public scoreboard
    client = await create_client("https://demo.ctfd.io")
    # For platforms requiring auth to see scoreboard, login first:
    # await client.auth.login(username="user", password="password")

    try:
        print("Fetching top 10 scoreboard entries...")
        # limit=0 would fetch all entries (if supported and not too large)
        top_10_entries = await client.scoreboard.get_top(limit=10)

        if top_10_entries:
            print("\n--- Top 10 Scoreboard ---")
            for entry in top_10_entries:
                print(f"  Rank: {entry.rank}, Name: {entry.name}, Score: {entry.score}")
        else:
            print("Scoreboard is empty or could not be fetched.")

        # Example: Fetching all scoreboard entries
        # print("\nFetching all scoreboard entries (limit=0)...")
        # all_entries = await client.scoreboard.get_top(limit=0)
        # if all_entries:
        #     print(f"Total entries on scoreboard: {len(all_entries)}")
        #     print(f"Top entry: Rank {all_entries[0].rank}, Name: {all_entries[0].name}, Score: {all_entries[0].score}")
        #     if len(all_entries) > 1:
        #         print(f"Last entry: Rank {all_entries[-1].rank}, Name: {all_entries[-1].name}, Score: {all_entries[-1].score}")
        # else:
        #     print("Full scoreboard is empty or could not be fetched.")

    except ScoreboardFetchError as e:
        print(f"Error fetching scoreboard: {e}")
    except UnauthorizedError:
        print("Error: Scoreboard access requires authentication on this platform.")
    except CTFInactiveError as e:
        print(f"Error: CTF or scoreboard is inactive: {e}")
    except CTFBridgeError as e:
        print(f"A CTFBridge error occurred: {e}")
    except Exception as e:
        print(f"An unexpected error occurred: {e}")


if __name__ == "__main__":
    asyncio.run(main())

---

Error Handling 💣

CTFBridge uses custom exceptions, all inheriting from CTFBridgeError. Catching these allows for more specific error management.

Always wrap your ctfbridge calls in appropriate try...except blocks for robust scripts!

Checking Platform Capabilities ✨

Different CTF platforms support different features. You can check what the initialized client supports before calling a function to avoid runtime errors and make your scripts more robust.

import asyncio

from ctfbridge import create_client
from ctfbridge.exceptions import CTFBridgeError


async def main():
    """
    This example demonstrates how to check the capabilities of a platform
    before attempting to use a feature. This allows you to write scripts
    that can adapt to different CTF platforms gracefully.
    """
    # Using CTFd, which supports all features
    client = await create_client("https://demo.ctfd.io")
    print(f"Client for {client.platform_name} at {client.platform_url} created.")

    print("\n--- Checking Platform Capabilities ---")

    # Check for login support (synchronous property access)
    if client.capabilities.login:
        print("✅ This platform supports login.")
        try:
            # Example of adapting behavior based on capability
            await client.auth.login(username="user", password="password")
            print("   -> Login successful!")
        except CTFBridgeError as e:
            print(f"   -> Login failed: {e}")
    else:
        print("❌ This platform does not support login via ctfbridge.")

    # Check for flag submission support
    if client.capabilities.submit_flags:
        print("✅ This platform supports flag submission.")
    else:
        print("❌ This platform does not support flag submission.")

    # Check for scoreboard support
    if client.capabilities.view_scoreboard:
        print("✅ This platform supports viewing the scoreboard.")
        try:
            top_entry = await client.scoreboard.get_top(1)
            if top_entry:
                print(f"   -> Top rank: {top_entry[0].name} with {top_entry[0].score} points.")
        except CTFBridgeError as e:
            print(f"   -> Could not fetch scoreboard: {e}")
    else:
        print("❌ This platform does not support viewing the scoreboard.")


if __name__ == "__main__":
    asyncio.run(main())