core

Downloading pages, turning them into markdown, crawling sites, and reading arXiv, GitHub, YouTube and PDF sources

This module holds everything fossick does with a single URL or file. fetch downloads a page and to_md turns its HTML into markdown. crawl follows links from one page to the next. find_xhr and paginate_api find the JSON API a page loads its data from and read it directly. Separate readers handle arXiv papers, GitHub repositories, YouTube videos and PDFs, and read, at the end of the notebook, picks the right one for any target.

---

source

fossick_cache

def fossick_cache(
    path:NoneType=None
):

Return the path path inside fossick’s cache folder (.fossick in the user cache directory), creating its parent folders.

---

source

Path.mk_bytes

def mk_bytes(
    data, mode:int=511, uid:int=-1, gid:int=-1
):

Write the bytes data to this path, creating any missing parent folders first.

---

source

syncy

def syncy(
    coro, tout:int=60
):

Run the coroutine coro from ordinary synchronous code and return its result; on timeout, cancel it and raise TimeoutError.

#: a timeout cancels the work and says it may already have happened; the coroutine's own `TimeoutError` passes through unchanged
async def _own(): raise TimeoutError('inner')
try: syncy(_own())
except TimeoutError as e: assert str(e) == 'inner'
else: raise AssertionError('no error')
async def _slow(): await asyncio.sleep(5)
try: syncy(_slow(), tout=0.1)
except TimeoutError as e: assert 'cancelled' in str(e) and 'may already' in str(e)
else: raise AssertionError('no timeout')

Turning HTML into markdown

A downloaded page is mostly markup that a reader does not need: navigation bars, scripts, styles and advertisements. to_md strips that away and returns the text as markdown. It accepts either a raw HTML string or a response returned by fetch, and with sel it converts only the part of the page that matches a CSS selector, such as sel='article'. html2md does the same for a string alone, and html_title returns a document’s title.

---

source

to_md

def to_md(
    res_or_html, # Page dict (from fetch/crawl) or raw HTML string
    sel:str=None, # CSS selector to extract before conversion; returns '' if no match
    multi:bool=False, # Return all selector matches joined
    wrap_tag:str=None, # Wrap each multi-result in <wrap_tag>...</wrap_tag>; only used when multi=True
    ignore_links:bool=True, ignore_images:bool=False,
    readability:bool=True, # False keeps the whole document, for local files and saved pages
)->str:

Convert a response from fetch or an HTML string to markdown, optionally keeping only the parts that match a CSS selector.

---

source

html2md

def html2md(
    s:str, ignore_links:bool=True, ignore_images:bool=False,
    readability:bool=True, # Extract the article (right for fetched pages); False converts the whole document
):

Convert the HTML string s to markdown, keeping only the main article unless readability=False.

---

source

html_title

def html_title(
    s:str
)->str:

Return the text of the document’s <title>, or of its first <h1> when it has no title, or an empty string.

By default to_md and html2md run readability, the article extractor behind browser reading modes. It finds the main article and drops everything around it, which is what you want for a page fetched from the web. It can also drop parts of a document you meant to keep, such as section headings. For a page someone saved on purpose, or a local HTML file, pass readability=False. The whole document is then converted, and only scripts, styles, noscript and template blocks and HTML comments are removed. The test below shows readability losing two <h3> headings that readability=False keeps.

Downloading pages

Refusing private addresses

A program that fetches URLs chosen by an AI agent can be made to fetch addresses inside your own network. Examples are a router’s admin page and a cloud server’s metadata service, which hands out credentials. This attack is called server-side request forgery. check_url refuses any URL that is not http or https. It refuses host names such as localhost and metadata.google.internal. It refuses any host that resolves to a private, loopback, link-local or other non-public address. It raises BlockedURL. The message names two ways to allow such a URL. Pass allow_private=True to one call, or set the environment variable FOSSICK_ALLOW_PRIVATE=1 for the whole process. get_page, fetch, http_get and http_post run this check first. http_page does not.

The check looks up the host name, and then the download looks it up again. The two lookups can get different answers. A person who controls the DNS for a name can give a public address to the check and 127.0.0.1 to the download. This attack is called DNS rebinding. To stop it, the plain HTTP path of get_page gives libcurl the addresses that the check approved, with the libcurl option CURLOPT_RESOLVE. libcurl then connects only to those addresses and does not look up the name again. The plain path also follows redirects itself. Before it connects to a new URL, it checks and pins the address of that URL in the same way. It does not send cookies, auth or credential headers that you supplied to a different host.

Some paths cannot pin the address, and keep only the first check. Through a proxy, also one that is set in the environment, the proxy looks up the name. The browser paths (heavy, stealthy, session, and auto=True after the plain step) check only the first URL. The browser does its own lookups, follows redirects, and loads scripts and images from other addresses with no check. For URLs that you do not trust, use the plain path.

---

source

check_url

def check_url(
    url:str, # the URL about to be fetched
    allow_private:bool=None, # True skips the check; None skips it only when FOSSICK_ALLOW_PRIVATE is 1, true, yes or on
)->str:

Return url unchanged when it is safe to fetch, or raise BlockedURL when it is not http or https or its host is private or reserved.

---

source

BlockedURL

def BlockedURL(
    *args, **kwargs
):

The error raised when a URL points at a private, local or cloud metadata address.

Downloading one page

get_page runs check_url, then downloads the page in one of four ways. By default it sends a plain HTTP request. heavy=True loads the page in a headless browser so that its JavaScript runs. stealthy=True uses scrapling’s stealth browser, which is built to pass bot checks such as Cloudflare’s. session=True uses the logged-in Chrome described in fossick.cdp, with its cookies. The browsers load on first use. When they cannot load, get_page raises BrowserUnavailable.

http_get and http_post are get_page with the method set and a 30 second timeout. All of them verify TLS certificates by default. Pass verify=False only for a host you trust that has a bad certificate. http_page sends a plain request with GET, POST, PUT or DELETE and skips the address check. Use http_page only for URLs your own code chose. Most callers need fetch, below.

Two optional limits apply to the plain path. max_bytes stops the download when the body becomes larger than that number of bytes, and raises TooLarge. A server that sends gigabytes then cannot fill the memory before to_md runs. max_recv_speed sets the highest download speed, in bytes per second. Both limits are off by default. The browser paths do not use them. http_page also takes max_bytes, and curl_options for other libcurl options.

Some networks inspect TLS traffic with their own root certificate. curl_cffi does not use the system keychain, so on such a machine a check of the certificate fails with curl: (60) ... unable to get local issuer certificate. Give the plain path a CA bundle file that holds the normal roots and the inspection root. Pass ca_bundle= or verify= with the path of the file, or set one of the environment variables SSL_CERT_FILE, CURL_CA_BUNDLE or REQUESTS_CA_BUNDLE. fossick reads them in that order, on each request. It uses the first one that names a file. On macOS, when none of these is set, fossick uses the system keychain. It reads the certificates of the system keychains once with security find-certificate, adds them to certifi’s roots, and keeps the result in its cache folder for one day. An inspection root that your company put in the system keychain then works with no setup. The bundle applies only when verification is on: with verify=True, as http_get sets it, or with ca_bundle=. fossick never turns verification off. On other systems, when none of these is set, nothing changes.

---

source

get_page

def get_page(
    url, method:str='GET', payload:NoneType=None, heavy:bool=False, stealthy:bool=False, session:bool=False,
    allow_private:NoneType=None, max_bytes:int=None, max_recv_speed:int=None, **kw
)->scrapling.engines.toolbelt.custom.Response:

Download url with plain HTTP, a headless browser, the stealth browser or the logged-in debug Chrome, after checking it with check_url.

---

source

http_page

def http_page(
    url, method:str='GET', payload:NoneType=None, verify:bool=True,
    max_bytes:int=None, # stop with `TooLarge` when the body is larger than this many bytes
    curl_options:dict=None, # extra libcurl options for this request, keyed by `curl_cffi.CurlOpt`
    ca_bundle:str=None, # CA bundle file to verify TLS with (default: $SSL_CERT_FILE, $CURL_CA_BUNDLE, $REQUESTS_CA_BUNDLE, then curl's own)
    **kw
)->scrapling.engines.toolbelt.custom.Response:

Send one plain HTTP request with scrapling’s Fetcher and return the response, without the private address check.

---

source

TooLarge

def TooLarge(
    *args, **kwargs
):

The error raised when a plain HTTP response body is larger than max_bytes.

---

source

BrowserUnavailable

def BrowserUnavailable(
    *args, **kwargs
):

The error raised when scrapling’s browser fetchers fail to import; the message suggests running scrapling install.

r1=get_page('https://httpbin.org/get', verify=False)
r1.__dict__.keys()

from fastcore.test import test_fail

try: get_page('https://httpbin.org/put', method='PUT', verify=False)
except AssertionError as e: test_fail(str(e) == 'Only GET and POST methods are supported')

http_page('https://httpbin.org/put', method='PUT', verify=False).status

fetch

Most code downloads a page with fetch. It wraps get_page and returns scrapling’s response object with two attributes added. html holds the page, or only the part matching sel. xhr holds the background requests recorded when capture_xhr=True. The response also has status, url, and a json() method for JSON replies.

auto=True tries the download methods in turn: plain HTTP, then the headless browser, then the stealth browser, then the stealth browser through the logged-in Chrome. A bot wall is a page that asks for a verification step instead of showing the requested content. auto=True stops at the first response that is not a bot wall. It records the method that worked in tier, or 'blocked' if none did. It records each failed attempt and its reason in errs.

browser_session downloads many pages with one browser. Starting a browser takes seconds. browser_session opens one browser for the length of a with block and gives you a fetch-like function that reuses it.

auto=True can end in the logged-in Chrome. That last method connects to the debug Chrome on port 9223, starts it if it is not running, and sends the cookies that Chrome has for the site. This is correct for a page that you want to read as yourself. It is not correct for a URL that comes from a web page, a document or another person. Such a URL can make your logged-in browser open a page with your cookies. When you do not trust the URL, pass auto=False. This is the default for fetch, but not for read. With auto=False, and without heavy, stealthy, session or capture_xhr, no function in this module starts a browser or connects to port 9223. This includes get_pdf and its proof-of-work step, read_arxiv, read_gh_file, crawl and read. The plain path always checks redirects: follow_redirects is 'safe' by default, and fossick checks and pins each step itself. The section Fetching untrusted URLs, near the end of this notebook, tests this.

---

source

browser_session

def browser_session(
    stealthy:bool=False, headless:bool=True, **init_kw
):

Keep one browser open for a with block and yield a fetch(url, sel=None) function that reuses it.

---

source

fetch

def fetch(
    url:str, # URL to fetch
    sel:str=None, # CSS selector to extract (None = full page)
    method:str='GET', # HTTP method; 'POST' sends payload as JSON body
    payload:dict=None, # POST body (JSON) or GET query params
    heavy:bool=False, # Full JS rendering via headless browser
    stealthy:bool=False, # Anti-bot stealth fetcher (Cloudflare etc.)
    capture_xhr:bool=False, # Capture XHR calls made by the page (only works with heavy or stealthy)
    session:bool=False, # Route through the persistent debug Chrome (reuses its logged-in cookies); True or a port int
    auto:bool=False, # Try plain, heavy, stealthy, then the logged-in debug Chrome with its cookies for the site; keep False for untrusted URLs
    allow_private:bool=None, # Allow localhost/intranet targets (default: refuse; $FOSSICK_ALLOW_PRIVATE=1)
    max_bytes:int=None, # Plain HTTP only: raise `TooLarge` when the body is larger than this many bytes (default: no limit)
    max_recv_speed:int=None, # Plain HTTP only: the highest download speed, in bytes per second (default: no limit)
    **kw
)->scrapling.engines.toolbelt.custom.Response: # Extra kwargs passed to scrapling (e.g. verify, headers, timeout)

Download url and return the response with .html set to the page, or the part matching sel, and .xhr set to any recorded background requests. auto=True can end in the logged-in debug Chrome, so pass auto=False for untrusted URLs.

Bot walls

Many sites put a challenge page in front of their content for visitors they think are scripts. Cloudflare’s “Just a moment” page and the Anubis proof-of-work page are common examples. The challenge often arrives with status 200, so it looks like a successful download. fossick treats a response as blocked when its status is 401, 403, 429, 503 or 530, when its HTML contains the marker of a known challenge or captcha widget, or when it is a 200 response with less than 200 characters of HTML. The tests below check these rules, including that a page which only mentions the word captcha is not counted as blocked.

The cells after that run against real sites and are not run by the tests. They show a Cloudflare and an Anubis wall being detected, auto=True stopping at the plain tier for a page that needs nothing more, browser_session reusing one browser for two pages, and session=True going through the logged-in Chrome.

# A site behind Cloudflare: a plain fetch hits the bot wall, and _blocked() flags it for escalation
cf = fetch('https://www.scrapingcourse.com/cloudflare-challenge', verify=False)
assert cf.status == 403, f'expected a Cloudflare 403 challenge, got {cf.status}'
assert _blocked(cf) is True            # fetch(auto=True) escalates from here (heavy -> stealthy -> session)
print('cloudflare:', cf.status, '| blocked?', _blocked(cf))

# A site behind Anubis (proof-of-work bot wall): _blocked() now recognizes it, so fetch(auto=True) escalates
an = fetch('https://anubis.techaro.lol/', verify=False)
assert 'making sure' in an.html.lower(), 'expected the Anubis challenge page'
assert _blocked(an) is True            # Anubis markers are in _BLOCK_MARKERS -> auto escalates past this wall
print('anubis wall detected by _blocked?', _blocked(an))

# regression: a normal page that merely *mentions* captcha must NOT read as a bot wall.
# Wikipedia embeds hCaptcha config keys in its page JS -- the word is there, but there's no challenge widget.
wiki = fetch('https://en.wikipedia.org/wiki/Anubis', verify=False)
assert wiki.status == 200 and 'captcha' in wiki.html.lower()   # the word IS present on the page...
assert _blocked(wiki) is False   # ...but _BLOCK_MARKERS matches captcha *widgets* (g-recaptcha/h-captcha/...), not the word

# fetch(auto=True): walks plain->heavy->stealthy->session, stops at the first unblocked tier, tags .tier
plain = fetch('https://example.com', auto=True, verify=False)
assert plain.status == 200 and plain.tier == 'plain'   # a static page needs no escalation
print('example.com ->', plain.tier)

# browser_session(): open ONE browser, reuse it across fetches (skips per-URL relaunch)
with browser_session() as bf:
    a = bf('https://example.com')
    b = bf('https://httpbin.org/html')
assert a.status == 200 and b.status == 200
assert len(a.html) > 0 and len(b.html) > 0

# fetch(session=True): route through the persistent debug Chrome, reusing its logged-in cookies
pg = fetch('https://httpbin.org/headers', session=True)
assert pg.status == 200
print(pg.html[:200])

Crawling and batches

crawl visits pages breadth-first, starting from one URL and following the links that match follow_sel, until it has visited max_pages pages. By default it stays on the starting site. It returns the responses that came back with status 200, and a page that fails to download is reported and skipped. delay waits between requests so as not to load a small site heavily. For crawls with heavy or stealthy, one browser is kept open for the whole crawl unless reuse=False.

---

source

crawl

def crawl(
    start_url:str, # URL to start from
    sel:str=None, # CSS selector to extract per page
    follow_sel:str='a[href]', # CSS selector for links to follow
    same_domain:bool=True, # Only follow links on same domain
    max_pages:int=10, # Max pages to visit
    delay:float=0, # Seconds to wait between requests (polite crawling)
    heavy:bool=False, # Full JS rendering via headless browser
    stealthy:bool=False, # Anti-bot stealth fetcher (Cloudflare etc.)
    reuse:bool=True, # For heavy/stealthy crawls, keep one browser open across pages
    **kw
)->list: # Extra kwargs passed to scrapling (e.g. verify, timeout)

Visit pages breadth-first from start_url, following links that match follow_sel, and return the responses that had status 200.

fetch_all downloads a list of URLs you already have, in parallel threads, and returns the responses in the same order. get_options reads the choices out of a <select> drop-down, which is how many sites list chapters, categories or years; the example uses a small page with three options.

---

source

fetch_all

def fetch_all(
    urls:list, # URLs to fetch
    sel:str=None, # CSS selector to extract per page (None = full page)
    concurrency:int=8, # Max parallel fetches
    heavy:bool=False, # Full JS rendering via headless browser
    stealthy:bool=False, # Anti-bot stealth fetcher (Cloudflare etc.)
    **kw
)->list: # Extra kwargs passed to fetch()

Fetch a list of URLs in parallel threads and return the responses in the same order.

---

source

get_options

def get_options(
    page_or_html, # Page dict (from fetch) or raw HTML string
    sel:str, # CSS selector for the <select> element
)->list:

Return the options of the <select> element matched by sel as a list of {'value', 'text'} dictionaries.

_sel_html = '''<html><body>
<select id="kanda">
  <option value="1">Balakanda</option>
  <option value="2">Ayodhyakanda</option>
  <option value="3">Aranyakanda</option>
</select>
</body></html>'''

opts = get_options(_sel_html, '#kanda')
assert opts == [{'value': '1', 'text': 'Balakanda'},
                {'value': '2', 'text': 'Ayodhyakanda'},
                {'value': '3', 'text': 'Aranyakanda'}]

PDFs and arXiv papers

get_pdf downloads a PDF, or reads one from disk, and returns it as a PdfDocument from pdf_oxide, or None when the bytes are not a PDF. It checks for the %PDF- signature in the first 1024 bytes of the content and ignores the URL. arXiv serves papers from URLs with no .pdf ending. The NCBI PubMed Central archive puts a small proof-of-work check in front of its PDFs. When the reply is such a check, get_pdf solves it and downloads the PDF again with the answer. Any page can contain the markers of this check, so the second download goes through the plain path of get_page, with the same address and redirect checks, timeout and max_bytes as the first. max_bytes limits the size of the PDF. When save_path is a folder or a file name, it saves a copy of the file. The examples download two papers from the Self-Determination Theory site and print the start of their text.

---

source

get_pdf

def get_pdf(
    url_or_path:str, # URL or local path to PDF
    save_path:NoneType=None, # optional path to save the PDF (directory or full path)
    max_bytes:int=None, # raise `TooLarge` when the download is larger than this many bytes (default: no limit)
    **scrapling_kw
)->pdf_oxide.pdf_oxide.PdfDocument | None: # extra kwargs passed to `get_page` (e.g. verify, headers, timeout, allow_private)

Download a PDF from a URL, or read one from disk, and return it as a PdfDocument, or None when the bytes are not a PDF.

import tempfile

sp = Path(tempfile.mkdtemp())

sdt = 'https://selfdeterminationtheory.org/SDT/documents/2000_RyanDeci_SDT.pdf'
sdt_p = get_pdf(sdt, save_path=sp, verify=False)

sdt_p.to_markdown_all()[:200]

gdr = 'https://selfdeterminationtheory.org/wp-content/uploads/2020/10/1997_GrolnickDeciRyan.pdf'
gcr = get_pdf(gdr, save_path=sp, verify=False)

gcr.to_markdown_all()[:200]

read_arxiv takes an arXiv URL, of either the abstract page or the PDF, or a bare id such as 2306.14881. It returns the paper’s title, authors, summary, published date, link, pdf_url and pdf_path. By default it also saves the PDF in save_dir. With source=True it adds the full text of the paper as markdown under source. The metadata is cached on disk. A second call for the same paper makes no request. force=True fetches the metadata again.

---

source

read_arxiv

def read_arxiv(
    url:str, # arxiv PDF URL, or arxiv abstract URL, or arxiv ID
    save_pdf:bool=True, # if True, saves the downloaded PDF to disk
    save_dir:str='.', # directory in which to save the PDF
    source:bool=False, # if True, also extract the paper's full text into `source`
    force:bool=False, # if True, forces re-download even if cached
    **kw
)->dict: # http_get kwargs (e.g. verify=False to skip SSL verification, headers=... etc.)

Return an arXiv paper’s metadata and PDF path, and optionally its full text, caching the metadata on disk.

read_arxiv('https://arxiv.org/abs/2306.14881')['summary'][:200]

lookup_doi finds the DOI of a paper from its title by asking Crossref, and returns it as a https://doi.org/ link, or None when Crossref has no match.

---

source

lookup_doi

def lookup_doi(
    title:str, # Paper title to search
)->str | None:

Return the doi.org URL of the first Crossref result for title, or None.

lookup_doi('Self-Determination Theory and the Facilitation of Intrinsic Motivation, Social Development, and Well-Being')

These functions turn documents into markdown or notebooks. pdf2md converts a PDF to one markdown string, with pages separated by ---, using pdflite. It saves images from the PDF in an images folder inside a folder named after out_path. This module re-exports pdflite’s own functions, such as pdf_md and clean_md. url2nb saves a web page, PDF or arXiv paper as a Jupyter notebook. pdf2nb saves a PDF as a notebook. Both return the notebook’s path. Both skip the work when the notebook already exists, unless force=True.

---

source

pdf2nb

def pdf2nb(
    u_or_p, # URL or local path to PDF
    nb_path:NoneType=None, # optional path to save the notebook; defaults to <pdf-stem>.ipynb
    force:bool=False, # if True, forces re-download and conversion even if notebook exists
    ocr_selection:str='auto', # choosing ocr (auto, off, on). auto uses pdf-oxide's ocr detection
    *, sel:str=None, method:str='GET', payload:dict=None, heavy:bool=False, stealthy:bool=False,
    capture_xhr:bool=False, session:bool=False, auto:bool=False, allow_private:bool=None, max_bytes:int=None,
    max_recv_speed:int=None
)->pathlib.Path: # extra kwargs passed to fetch() (e.g. verify=False to skip SSL verification)

Save a PDF as a Jupyter notebook and return the notebook’s path.

---

source

url2nb

def url2nb(
    url, # URL to convert (PDF, arxiv, or HTML page)
    nb_path:NoneType=None, # optional path to save the notebook; defaults to <url-stem>.ipynb
    force:bool=False, # if True, forces re-download and conversion even if notebook exists
    *, sel:str=None, method:str='GET', payload:dict=None, heavy:bool=False, stealthy:bool=False,
    capture_xhr:bool=False, session:bool=False, auto:bool=False, allow_private:bool=None, max_bytes:int=None,
    max_recv_speed:int=None
)->pathlib.Path: # extra kwargs passed to fetch() (e.g. verify=False to skip SSL verification)

Save a web page, PDF or arXiv paper as a Jupyter notebook and return the notebook’s path.

---

source

pdf2md

def pdf2md(
    pdf, # a PdfDocument, path, bytes or URL
    out_path:str | pathlib.Path, # notebook path; images go under its stem
    image_dir:str='images', *, pages:bool=False, ocr:str='auto', preserve_layout:bool=False, **kw
)->str: # forwarded to pdflite.pdf_md

Convert a PDF to one markdown string with pages separated by ---, saving its images beside out_path.

get_pdf('https://papers.neurips.cc/paper/7181-attention-is-all-you-need.pdf', save_path=sp, verify=False)

url2nb('https://squiddev.medium.com/continuing-continuations-cps-in-python-47bba90c8d1e', nb_path=sp, verify=False)

pdf2nb(sdt, nb_path=Path(sp)/'test.ipynb', verify=False)

import tempfile
from pdf_oxide import Pdf

_c = _md2cells('# Intro\n\nSome prose.\n\n[code]\n\n> 1+1\n2\n[/code]\n\n## Next\n\nMore.')
assert [(c.cell_type, c.source) for c in _c] == [('markdown','# Intro\n\nSome prose.'), ('code','1+1\n# => 2'), ('markdown','## Next\n\nMore.')], _c
assert _md2cells('[code]\n\ndef f(x):\nreturn x\n[/code]')[0].source == 'def f(x):\n    return x'
_t = Path(tempfile.mkdtemp()); Pdf.from_markdown('# Attention\n\nSome prose about attention.').save(str(_t/'a.pdf'))
_rf, fetch = fetch, lambda url, **kw: '<html><body><h1>Title</h1><p>' + 'Some prose about notebooks. '*20 + '</p></body></html>'
try: _hp = url2nb('https://example.com/page', nb_path=_t/'page.ipynb')
finally: fetch = _rf
assert 'Some prose about notebooks' in _hp.read_text()
_seen, _rp = [], pdf_md
def pdf_md(*a, **kw): _seen.append(kw.get('ocr')); return _rp(*a, **kw)
try: _pp = pdf2nb(str(_t/'a.pdf'), nb_path=_t/'attn.ipynb', ocr_selection='off')
finally: pdf_md = _rp
assert _seen == ['off'] and 'Attention' in _pp.read_text()

GitHub

gh_clone returns a local copy of a repository. The first call clones it into fossick’s cache folder, in a folder named owner__repo. It clones over HTTPS only, and never asks for a password. Pass ssh=True to try SSH after HTTPS, for a private repository and an SSH key. SSH then runs in batch mode, so it fails at once when it needs a passphrase. The clone is shallow: it gets only the latest commit. Pass depth=None for the full history, or a different number of commits. timeout is the time limit for each git command, in seconds. When git takes longer, fossick stops git, removes the part-made folder and raises TimeoutError. Later calls fetch the clone and reset it to the latest commit. Git runs with /dev/null as stdin and never reads the caller’s terminal. fossick gives git its own environment and does not change os.environ. fossick refuses a target that starts with -, and puts -- before the URL, so that git cannot read a URL as an option. gh_clone returns a local path unchanged. Code that accepts a repository therefore accepts a folder on disk. read_gh_repo reads files from that clone and returns {path: text}. By default it reads the README, pyproject.toml and every Python file. It caches its results for an hour. read_gh_file downloads one file as raw text from its GitHub page URL. It returns source code exactly as written. It converts the file to markdown only when GitHub serves HTML.

---

source

read_gh_file

def read_gh_file(
    url:str, # GitHub blob URL of the file to read
    **kw
):

Return the contents of one file from its GitHub page URL.

---

source

read_gh_repo

def read_gh_repo(
    path_or_url:str, # GitHub URL, SSH address, or local path
    globs:tuple=None, # file glob patterns (default: README*, pyproject.toml, *.py)
    limit:int=None, # max files to return
    as_list:bool=False, # return list of Paths instead of {path: content} dict
)->dict:

Return the files of a GitHub repository or local folder that match globs, as a {path: text} dictionary.

---

source

gh_clone

def gh_clone(
    path_or_url:str, # GitHub URL, SSH address, or a local path (returned as-is)
    depth:int=1, # commits of history to clone; None for all of them
    timeout:float=300, # seconds for each git command; then git is stopped and the part-made folder removed
    ssh:bool=False, # try SSH after HTTPS, for a private repo and an SSH key
)->pathlib.Path:

Return a local clone of a GitHub repository, cloned or updated in fossick’s cache, or return a local path unchanged.

These tests give gh_clone a fake subprocess.run, so they need no network. By default the clone is HTTPS only and shallow, and -- comes before the URL. A failed HTTPS clone does not try SSH, because SSH can ask for a passphrase. With ssh=True, SSH comes second and runs in batch mode. Git gets its own environment, and os.environ stays the same. A clone that passes timeout raises TimeoutError and leaves no folder. A target that starts with - is refused before git runs.

from unittest.mock import patch
import uuid

_runs = []
def _fake_git(fail=(), slow=False):
    def run(cmd, **kw):
        _runs.append((cmd, kw))
        if cmd[1] == 'clone':
            d = Path(cmd[-1]); d.mkdir(parents=True, exist_ok=True); (d/'half').write_text('x')
            if slow: raise subprocess.TimeoutExpired(cmd, kw['timeout'])
            if any(f in cmd[-2] for f in fail): return subprocess.CompletedProcess(cmd, 128, '', 'fatal: no access')
        return subprocess.CompletedProcess(cmd, 0, '', '')
    return run

_url = f'https://github.com/fossick-test/{uuid.uuid4().hex}'
_d = _clone_dir(_gh_remotes(_url)[0])
_before = dict(os.environ)
with patch('subprocess.run', _fake_git(fail=('https://',))):
    test_fail(lambda: gh_clone(_url), contains='no access')
test_eq(len(_runs), 1)                                        # HTTPS only: no SSH fallback
cmd, kw = _runs[0]
test_eq(cmd, ['git', 'clone', '--depth', '1', '--', _gh_remotes(_url)[0], str(_d)])
test_eq(kw['env']['GIT_TERMINAL_PROMPT'], '0'); test_eq(kw['timeout'], 300); test_eq(kw['stdin'], subprocess.DEVNULL)
assert not _d.exists()                                        # the failed clone left nothing behind
test_eq(dict(os.environ), _before)                            # os.environ is never changed

_runs.clear()
with patch('subprocess.run', _fake_git(fail=('https://',))): test_eq(gh_clone(_url, depth=None, ssh=True), _d)
test_eq([c[2:] for c, _ in _runs], [['--', _gh_remotes(_url)[0], str(_d)], ['--', _gh_remotes(_url, ssh=True)[1], str(_d)]])
assert _gh_remotes(_url, ssh=True)[1].startswith('git@github.com:')
assert 'BatchMode=yes' in _runs[1][1]['env']['GIT_SSH_COMMAND']
shutil.rmtree(_d)

_runs.clear()
with patch('subprocess.run', _fake_git(slow=True)): test_fail(lambda: gh_clone(_url, timeout=5), contains='more than 5')
test_eq(_runs[0][1]['timeout'], 5)
assert not _d.exists()                                        # stopped git, removed the part-made folder

_runs.clear()
with patch('subprocess.run', _fake_git()):
    test_fail(lambda: gh_clone('--upload-pack=touch /tmp/x'), contains='starts with "-"')
test_eq(_runs, [])
test_eq(_gh_remotes('git@github.com:a/b.git'), ['https://github.com/a/b.git'])   # an SSH address clones over HTTPS too

read_gh_file('https://github.com/Karthik777/litesearch/blob/main/README.md')[:200]

list(read_gh_repo('https://github.com/vedicreader/gheasy'))

Finding the API behind a page

Many modern pages arrive nearly empty. JavaScript then loads their data with background requests, called XHR or fetch calls, which usually return JSON. You can read that JSON directly instead of scraping the rendered page. find_xhr loads a page in a headless browser, records the background requests it makes, and returns each one whose URL matches pattern as a dictionary with url, content_type and the decoded data. With session=True it records the requests through the logged-in Chrome instead, and each result also carries a capture. replay_xhr sends a captured request again as a plain HTTP call with the browser’s cookies. Use it to call an API that needs a login without keeping a browser open.

pattern can be a regular expression, such as .*products.*, or a shell-style glob, such as *products*. compile_pattern accepts both: it tries the text as a regular expression and, when that is invalid, reads it as a glob.

---

source

replay_xhr

def replay_xhr(
    capture:dict, # a request dict from cdp.calls() or find_xhr(session=True)[i].capture
    data:NoneType=None, # override request body (str/dict); defaults to the captured body
    port:int=9223, # debug Chrome port to pull cookies from
    use_cookies:bool=True, # attach the browser's cookies for this domain (keeps auth)
    **kw
)->scrapling.engines.toolbelt.custom.Response: # Extra kwargs passed to the underlying Fetcher (headers, params, verify)

Send a request captured from the browser again as a plain HTTP call, with the debug Chrome’s cookies for that site.

---

source

find_xhr

def find_xhr(
    url:str, # URL to visit with browser
    pattern:str='*', # Glob or regex pattern to filter captured XHR URLs
    session:bool=False, # Use the persistent debug Chrome (authenticated) instead of a throwaway browser
    port:int=9223, # debug Chrome port (only when session=True)
    tail:int=3, # seconds to keep listening after navigation (only when session=True)
    **kw
)->list: # Extra kwargs passed to fetch (verify, network_idle, etc.)

Load url in a browser and return the background requests it makes whose URLs match pattern.

---

source

compile_pattern

def compile_pattern(
    pattern
):

Compile pattern as a regular expression, or as a shell-style glob such as *foo* when it is not a valid one.

assert compile_pattern('.*products.*').search('https://api.example.com/products?q=1')
assert compile_pattern('*products*').search('https://api.example.com/products?q=1')
assert not compile_pattern('*products*').search('https://api.example.com/search')
assert compile_pattern('.*[Ss]earch.*').search('https://api.example.com/Search')

An API rarely returns its records at the top level of the JSON. They might be under results, or several levels down at data.products.items. json_records finds them by searching the whole payload for the longest list of dictionaries, and with_key=True also tells you which key held it.

paginate_api reads every page of a paged JSON API. It sends the request repeatedly, raising the page number in page_field each time, and stops when a page comes back empty or shorter than the page size, or after max_pages pages. It finds each page’s records with results_field if you name one and json_records if you do not. With save=True each page is written to a file, and a later run reads the saved pages instead of fetching them again.

---

source

paginate_api

def paginate_api(
    url:str, # API endpoint URL
    payload:dict=None, # Request body (POST) or params (GET)
    page_field:str='pageNumber', # Payload key to increment for each page
    size_field:str='pageSize', # Payload key for page size (detects last page)
    results_field:str=None, # Response key with items list (auto-detect if None)
    method:str='POST', # HTTP method
    max_pages:int=10, page_size:int=24, # Page size to request (only used if not in payload)
    page_start:int=1, # Starting page number (default 1)
    save:bool=False, # If True, saves each page's items to disk
    save_file:str='{url}_page_{page}.json', # Filepath pattern for saving (only used if save=True)
    force:bool=False, # If True, forces re-fetching even if saved file exists
    **kw
)->list: # Extra kwargs passed to fetch() (verify, headers, etc.)

Request successive pages of a JSON API and return the items from every page in one list.

---

source

json_records

def json_records(
    data, # any decoded JSON payload
    min_len:int=2, # shortest list of dicts worth calling records
    key:str=None, # only accept a list held under this key, at any depth
    with_key:bool=False, # also return the key the records were found under
    max_depth:int=6
):

Return the longest list of dictionaries found anywhere in a decoded JSON payload.

download_files saves every file linked from a page whose link matches pattern, for example *.pdf or .*\.ipynb$. With follow=True it first crawls from the page, so it can collect files spread over a series of pages. Files already on disk are not downloaded again unless force=True. The example walks the JAX tutorial pages by their “next” links and saves each notebook.

---

source

download_files

def download_files(
    url:str, pattern:str='*', # Glob or regex to match download hrefs
    save_dir:str='.', # Directory to save files
    follow:bool=False, # Crawl linked pages looking for more files
    follow_sel:str='a[href]', # Selector for page links to follow when crawling
    file_sel:str='a[href]', # Selector to scan for download links on each page
    same_domain:bool=True, max_pages:int=10, delay:float=0,
    force:bool=False, # Re-download even if file already exists
    heavy:bool=False, stealthy:bool=False, **kw
)->list[pathlib.Path]: # Extra kwargs passed to fetch/http_get (e.g. verify)

Download every file linked from url whose link matches pattern, optionally crawling linked pages first, and return the saved paths.

jax, fs = 'https://docs.jax.dev/en/latest/notebooks/thinking_in_jax.html', 'a[class="right-next"]'
download_files(jax, follow=True, follow_sel=fs, file_sel='a[href$=".ipynb"]', save_dir=sp, verify=False)

from fastcore.test import test_eq

_pages = crawl('https://lego.sankalpa.sh/', max_pages=2, verify=False)
assert isinstance(_pages, list), f"Expected list, got {type(_pages)}"
assert len(_pages) > 0, "Expected at least one page"
assert all(p.status == 200 for p in _pages), "Non-200 pages should be skipped"
assert all(len(p.html) > 0 for p in _pages), "html should be non-empty"
assert len({p.url for p in _pages}) == len(_pages), "url values should be unique"

# crawl(reuse=True): a JS-rendered crawl keeps ONE browser open across pages (no per-page relaunch)
_jsp = crawl('https://quotes.toscrape.com/js/', max_pages=2, heavy=True, reuse=True)
assert _jsp and all(p.status == 200 for p in _jsp)
assert all(len(p.html) > 500 for p in _jsp)      # JS actually rendered content into the DOM

# Step 1: visit the listing page with a headless browser, capture all XHR/fetch calls
apis = find_xhr('https://www.danmurphys.com.au/list/wine-all', verify=False)

# paginate_api: test with JSONPlaceholder (free public REST API, GET-based)
posts = paginate_api(
    'https://jsonplaceholder.typicode.com/posts',
    payload={'_page': 1, '_limit': 5},
    page_field='_page',
    size_field='_limit',
    method='GET',
    verify=False,
)
assert len(posts) >= 5
assert 'title' in posts[0]

find_xhr(session=True) records requests through the logged-in Chrome, so a request that needs your cookies can be captured and then replayed with replay_xhr:

# find_xhr(session=True) -> replay_xhr: capture a request via the authenticated debug Chrome,
# then replay it as a fast plain-HTTP call using the browser's cookies
hits = find_xhr('https://httpbin.org/json', pattern='.*json', session=True)
assert hits, 'no matching request captured'
assert hits[0].capture, 'each hit should carry a replayable capture'
r = replay_xhr(hits[0].capture)
assert r.status == 200
print(hits[0].url, '->', r.status)

YouTube

These three functions use yt-dlp. search_yt searches YouTube and returns a list of result dictionaries. read_yt returns one video’s details and its transcript under the key source. The transcript comes from YouTube’s English automatic captions. A video without them has an empty transcript. read_yt(url, cues=True) adds cues, a list of {start, end, text} with times in seconds, so a caller can seek to a line. yt_link(url, seconds) makes a watch link that opens at that time. read_yt caches results, cues included, on disk by video id. download_yt saves a video’s audio as mp3 or its video as mp4. The mp3 format needs ffmpeg. All three accept verify=False to skip certificate checks. Use it behind a proxy that rewrites HTTPS.

---

source

yt_link

def yt_link(
    url_or_id:str, # YouTube URL or video ID
    seconds:float, # moment to open the video at
)->str:

Return a watch link that opens the video at seconds.

---

source

search_yt

def search_yt(
    q:str, # search query
    n:int=10, # number of results to return
    **kw
)->list: # extra opts forwarded to YoutubeDL; verify= is translated to nocheckcertificate, proxy= passed as-is

Search YouTube with yt-dlp and return up to n results as dictionaries.

results = search_yt('3blue1brown neural networks', n=3)
assert len(results) >= 1, f"expected results, got {len(results)}"
assert any(kw in results[0]['title'].lower() for kw in ('3blue1brown', 'neural', 'network')), \
    f"unexpected title: {results[0]['title']}"
assert results[0]['url'].startswith('https://www.youtube.com'), f"bad url: {results[0]['url']}"
print(results[0]['title'], '|', results[0]['url'])

---

source

read_yt

def read_yt(
    url:str, # YouTube URL or video ID
    force:bool=False, # if True, forces re-fetching even if cached
    cues:bool=False, # if True, add `cues`: `[{start, end, text}]` in seconds
    **kw
)->dict: # SSL/proxy opts (e.g. **ddgs_env()); verify= translated to nocheckcertificate

Return a YouTube video’s title, channel, description, duration, upload date and English transcript, cached by video id. With cues=True it also gives timed transcript lines.

The cue parser is tested on a saved YouTube caption sample, so no network is needed. YouTube repeats each line in the next cue as the caption rolls up. The parser folds the repeats, so each spoken line shows once, with its start and end time.

from fastcore.test import test_eq, test_fail

_vtt = '''WEBVTT
Kind: captions
Language: en

00:00:00.160 --> 00:00:02.230 align:start position:0%
 
hello<00:00:00.640><c> and</c><00:00:00.880><c> welcome</c><00:00:01.440><c> to</c><00:00:01.600><c> the</c><00:00:01.760><c> show</c>

00:00:02.230 --> 00:00:02.240 align:start position:0%
hello and welcome to the show
 

00:00:02.240 --> 00:00:04.470 align:start position:0%
hello and welcome to the show
today<00:00:02.800><c> we</c><00:00:03.040><c> learn</c><00:00:03.520><c> about</c><00:00:03.919><c> neurons</c>

00:00:04.470 --> 00:00:04.480 align:start position:0%
today we learn about neurons
 

00:00:04.480 --> 00:00:07.109 align:start position:0%
today we learn about neurons
each<00:00:04.799><c> neuron</c><00:00:05.200><c> holds</c><00:00:05.600><c> a</c><00:00:05.759><c> number</c>

00:00:07.109 --> 00:00:07.119 align:start position:0%
each neuron holds a number
 

00:01:05.000 --> 00:01:07.500
that is all for today
'''
cues = _parse_cues(_vtt)
test_eq([c['text'] for c in cues], ['hello and welcome to the show','today we learn about neurons','each neuron holds a number','that is all for today'])
test_eq([(c['start'], c['end']) for c in cues][:2], [(0.16, 2.23), (2.24, 4.47)])
test_eq((cues[-1]['start'], cues[-1]['end']), (65.0, 67.5))
assert all(isinstance(c['start'], float) and isinstance(c['end'], float) for c in cues)
test_eq(_parse_cues(''), [])
test_eq(_parse_cues('WEBVTT\n\n'), [])
test_eq(' '.join(c['text'] for c in cues), _parse_vtt(_vtt))

test_eq(yt_link('https://youtu.be/aircAruvnKk?t=3', 83.9), 'https://www.youtube.com/watch?v=aircAruvnKk&t=83s')
test_eq(yt_link('aircAruvnKk', 0), 'https://www.youtube.com/watch?v=aircAruvnKk&t=0s')
test_fail(lambda: yt_link('not-an-id', 5), contains='video id')
test_fail(lambda: yt_link('aircAruvnKk/../x', 5), contains='video id')

# read_yt with a stubbed yt-dlp and fetch: cues kept in the cache, hidden by default
class _YDL:
    def __init__(s, o): pass
    def __enter__(s): return s
    def __exit__(s, *a): pass
    def extract_info(s, u, download=False): return dict(title='t', automatic_captions={'en':[dict(ext='vtt', url='u')]})
class _Mod: YoutubeDL = _YDL
class _R: status=200; body=_vtt.encode()
_old = (globals()['_ytdl'], globals()['http_get'], globals()['_yt_index'])
_c = {}
try:
    globals().update(_ytdl=lambda: _Mod, http_get=lambda u: _R, _yt_index=lambda: _c)
    a = read_yt('aircAruvnKk')
    assert 'cues' not in a and a['source'].startswith('hello and welcome')
    b = read_yt('aircAruvnKk', cues=True)
    test_eq(b['cues'], cues); test_eq(b['source'], a['source'])
    assert 'cues' in _c['aircAruvnKk']
    _c['aircAruvnKk'].pop('cues')  # an old cache entry has no cues: refetch
    test_eq(read_yt('aircAruvnKk', cues=True)['cues'], cues)
    globals().update(http_get=lambda u: type('R',(),dict(status=404,body=b''))())
    e = read_yt('aircAruvnKk', cues=True, force=True)
    test_eq((e['source'], e['cues']), ('', []))
finally: globals().update(_ytdl=_old[0], http_get=_old[1], _yt_index=_old[2])

meta = read_yt('https://www.youtube.com/watch?v=aircAruvnKk')
assert meta['title'], "title should be non-empty"
assert isinstance(meta['source'], str), "source should be a string"
assert len(meta['source']) > 100, "transcript should be substantial"

---

source

download_yt

def download_yt(
    url:str, # YouTube URL or video ID
    format:str='audio', # 'audio'|'video'|yt-dlp format string
    save_dir:str='.', # directory to save file
    quality:str=None, # preferred audio quality (e.g. '192' for 192kbps, only used if format='audio')
    **kw
)->pathlib.Path: # SSL/proxy opts (e.g. **ddgs_env()); verify= translated to nocheckcertificate

Download a YouTube video’s audio as mp3, its video as mp4, or any yt-dlp format, and return the saved file’s path.

p = download_yt('https://www.youtube.com/watch?v=aircAruvnKk', format='audio', save_dir='/tmp/fossick_test')
assert p.exists(), f"file not found: {p}"
assert p.suffix == '.mp3', f"expected .mp3, got {p.suffix}"
print(f"saved to: {p} ({p.stat().st_size // 1024} KB)")

One function for every source

read accepts every kind of target the readers above handle and returns the same seven fields for all of them: ok, kind, title, source, text, skipped and meta. A program can treat web pages, PDFs, arXiv papers, YouTube videos, GitHub content and local files alike.

what_is decides which reader a target needs. A path that exists on disk wins over every URL rule, so a local file named pdf is read from disk rather than fetched. A GitHub URL containing /blob/ is one file (ghfile) rather than a whole repository (github). Any other http or https URL is a web page. A failed read returns ok=False with the reason in skipped. Only a target that is none of these kinds raises ValueError.

what_is reads the host of a URL to find arXiv and YouTube links, not the full text of the URL. https://docs.example.dev/youtube.com-guide is a web page, not a YouTube video.

---

source

what_is

def what_is(
    target:str, # a URL, an arXiv id, a YouTube link, a GitHub repo or file, a PDF, a path
)->str:

Return the name of the reader read would use for target.

---

source

md_title

def md_title(
    md:str, # markdown to take a title from
    fallback:str='', # what to use when it has no heading and no first line
)->str:

Return the first heading in md, else its first non-empty line, else fallback.

pdf_pages converts a PDF and splits the result at the page separators into (page_index, markdown) pairs, numbered from 0. read uses it when called with pages=True.

---

source

pdf_pages

def pdf_pages(
    pdf:pdf_oxide.pdf_oxide.PdfDocument, # an open PDF
    out_path:str | pathlib.Path, # where extracted images go, as `pdf2md` takes it
    **kw
)->list: # forwarded to `pdf2md`

Convert a PDF to markdown and return it as (page_index, markdown) pairs, numbered from 0.

---

source

read

def read(
    target:str, # a URL, an arXiv id, a YouTube link, a GitHub repo or file, a PDF, a local path
    sel:str=None, # CSS selector, for the web cases
    auto:bool=True, # escalate plain -> heavy -> stealthy -> logged-in Chrome past bot walls; pass False for untrusted URLs
    pages:bool=False, # PDFs as `[(page_index, text)]`, numbered from 0, rather than one string
    save_dir:str='.', # where a downloaded PDF is kept
    force:bool=False, # ignore whatever the reader has cached
    **kw
)->fastcore.basics.AttrDict: # `(ok, kind, title, source, text, skipped, meta)`

Read any supported target and return its text and details in one common shape.

# the network cases, which the CI run does not reach
r = read('https://arxiv.org/abs/1706.03762')
r.ok, r.kind, r.title, r.meta['authors'][:2]

p = read('https://papers.neurips.cc/paper/7181-attention-is-all-you-need.pdf', pages=True)
p.ok, p.kind, len(p.text), p.text[0][0]

Fetching untrusted URLs

A program that reads links from web pages, documents or messages fetches URLs that another person can choose. For these URLs, call fetch, get_pdf and read with auto=False. fossick then uses only plain HTTP. It checks and pins each address, and checks each redirect again. max_bytes can limit the size of each download.

The tests below use a small web server in this process, on 127.0.0.1. The address check refuses 127.0.0.1, so _public_loopback makes the check count it as public for the length of a test. The server can then act as a site on the internet. _public_loopback can also give a name an address that only the check sees. libcurl does not see that address.

Any page can contain the proof-of-work markers that get_pdf looks for. Here the page contains them, and the answer gets a redirect to localhost. get_pdf refuses the redirect. The server log shows that the request to /secret was not sent.

_Site.hits.clear()
with _public_loopback(): test_fail(lambda: get_pdf(f'{_base}/paper.pdf'), contains='localhost is an internal host')
test_eq(_Site.hits, ['/paper.pdf', '/paper.pdf'])   # the answer was sent; the redirect to /secret was not followed

fossick checks a redirect by the address that its host resolves to, not only by its name. Here inside.invalid resolves to 10.0.0.7, so fossick refuses the redirect. The second fetch tests the pin. Names that end in .invalid never resolve in real DNS. The fetch can reach the server only through the address that the check approved and gave to libcurl.

_Site.hits.clear()
with _public_loopback({'inside.invalid': '10.0.0.7'}):
    test_fail(lambda: fetch(f'{_base}/to-inside'), contains='resolves to 10.0.0.7')
with _public_loopback({'pinned.invalid': '127.0.0.1'}):
    r = fetch(f'http://pinned.invalid:{_srv.server_port}/page')
test_eq(r.status, 200)
test_eq(_Site.hits, ['/to-inside', '/page'])

max_bytes stops a large body. The server sends 4 MB with no Content-Length. fetch and get_pdf stop after 100,000 bytes and raise TooLarge. With no max_bytes, the download is not limited, as before.

test_fail(lambda: fetch(f'{_base}/big', allow_private=True, max_bytes=100_000, retries=1), contains='max_bytes=100000')
test_fail(lambda: get_pdf(f'{_base}/big', allow_private=True, max_bytes=100_000, retries=1), contains='max_bytes=100000')
with _public_loopback(): test_fail(lambda: fetch(f'{_base}/big', max_bytes=100_000, retries=1), contains='max_bytes')
test_eq(len(fetch(f'{_base}/big', allow_private=True).body), 64 * 65536)   # no limit unless one is asked for

A CA bundle from SSL_CERT_FILE is used on each request. The test starts an HTTPS server with a self-signed certificate, made with the openssl command. Without the bundle, the certificate check fails. With SSL_CERT_FILE set to the certificate, or with ca_bundle=, the download works. verify=True stays on in each case.

import ssl, tempfile
_td = Path(tempfile.mkdtemp())
subprocess.run(['openssl', 'req', '-x509', '-newkey', 'rsa:2048', '-nodes', '-days', '1', '-subj', '/CN=127.0.0.1',
                '-addext', 'subjectAltName=IP:127.0.0.1', '-keyout', str(_td/'key.pem'), '-out', str(_td/'cert.pem')],
               check=True, capture_output=True, stdin=subprocess.DEVNULL)
_tls = ThreadingHTTPServer(('127.0.0.1', 0), _Site)
_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH); _ctx.load_cert_chain(_td/'cert.pem', _td/'key.pem')
_tls.socket = _ctx.wrap_socket(_tls.socket, server_side=True)
threading.Thread(target=_tls.serve_forever, daemon=True).start()
_https = f'https://127.0.0.1:{_tls.server_port}/notes.txt'

_env = {v: os.environ.pop(v) for v in ('SSL_CERT_FILE', 'CURL_CA_BUNDLE', 'REQUESTS_CA_BUNDLE') if v in os.environ}
try:
    test_fail(lambda: http_get(_https, allow_private=True, retries=1), contains='(60)')
    test_eq(http_get(_https, allow_private=True, retries=1, ca_bundle=_td/'cert.pem').body, b'secret')
    os.environ['SSL_CERT_FILE'] = str(_td/'cert.pem')
    test_eq(http_get(_https, allow_private=True, retries=1).body, b'secret')
    with _public_loopback(): test_eq(fetch(_https, verify=True, retries=1).body, b'secret')   # the pinned path too
finally:
    os.environ.pop('SSL_CERT_FILE', None); os.environ.update(_env)
    _tls.shutdown(); _tls.server_close()

On macOS, _system_ca makes the CA bundle from certifi’s roots and the output of security find-certificate. The first test gives it a fake security command, so it runs on every system. The bundle holds the certifi roots and the fake certificate. A fresh file is used again without a new security call. When security gives no certificate, _system_ca keeps the old file. Off macOS, _system_ca returns None and curl uses its own roots.

from unittest.mock import patch
import certifi, tempfile

_fake_pem = '-----BEGIN CERTIFICATE-----\nZmFrZQ==\n-----END CERTIFICATE-----\n'
_calls = []
def _fake_run(cmd, **kw):
    _calls.append(cmd)
    return subprocess.CompletedProcess(cmd, 0, _fake_pem if cmd[0] == 'security' else '', '')
_ca = Path(tempfile.mkdtemp())/'ca.pem'
with patch.object(sys, 'platform', 'darwin'), patch('subprocess.run', _fake_run):
    test_eq(_system_ca(_ca), str(_ca))
    test_eq(_calls[0][:4], ['security', 'find-certificate', '-a', '-p'])
    txt = _ca.read_text()
    assert txt.startswith(Path(certifi.where()).read_text()) and txt.endswith(_fake_pem)
    _system_ca(_ca); test_eq(len(_calls), 1)                 # a fresh file is used again
    _fake_pem = ''
    test_eq(_system_ca(_ca, max_age=0), str(_ca))            # no certificates: keep the old file
    test_eq(_system_ca(Path(tempfile.mkdtemp())/'none.pem'), None)
with patch.object(sys, 'platform', 'linux'): test_eq(_system_ca(_ca), None)

On a Mac, the real bundle holds many certificates, and _verify gives it to curl when no CA variable is set. Verification stays on.

if sys.platform == 'darwin':
    _env = {v: os.environ.pop(v) for v in _CA_ENV if v in os.environ}
    try:
        _p = _verify()
        test_eq(_p, _system_ca())
        assert Path(_p).read_text().count('BEGIN CERTIFICATE') > 100
        test_eq(_verify(False), False)                        # verify=False still means off, as before
    finally: os.environ.update(_env)

With auto=False, no path starts a browser or connects to the debug Chrome. The test replaces the browser loaders, and the fossick.cdp functions that connect to port 9223, with functions that record each call. It then runs fetch, read for a web page and for PDFs, the proof-of-work step, crawl and read_gh_file. One of the pages is a bot wall. No call is recorded. As a control, fetch(auto=True) on the same bot wall calls both.

import fossick.cdp as _cdp
_touched = []
def _no_browser(*a, **k): _touched.append('browser'); raise BrowserUnavailable('no browser in this test')
def _no_cdp(*a, **k): _touched.append('cdp'); raise RuntimeError('no debug Chrome in this test')
_saved = _browser_fetchers, browser_session, _cdp.cdp_ws, _cdp.cdp_cookies, _cdp.cdp_connect
_browser_fetchers = browser_session = _no_browser
_cdp.cdp_ws = _cdp.cdp_cookies = _cdp.cdp_connect = _no_cdp
try:
    with _public_loopback():
        test_eq(fetch(f'{_base}/wall').status, 403)
        assert not read(f'{_base}/wall', auto=False).ok
        assert not read(f'{_base}/wall/doc.pdf', auto=False).ok
        test_fail(lambda: read(f'{_base}/paper.pdf', auto=False), contains='internal host')
        test_eq(len(crawl(f'{_base}/page', max_pages=3)), 1)
        test_eq(read_gh_file(f'{_base}/notes.txt'), 'secret')
        test_eq(_touched, [])
        _r = fetch(f'{_base}/wall', auto=True)   # the control: auto=True does try the browsers and the debug Chrome
finally: _browser_fetchers, browser_session, _cdp.cdp_ws, _cdp.cdp_cookies, _cdp.cdp_connect = _saved
test_eq(_r.tier, 'blocked')
assert {'browser', 'cdp'} <= set(_touched), _touched
_srv.shutdown(); _srv.server_close()

fossick: no browser in this test

Installing the agent skill

fossick includes a SKILL.md that tells coding agents how to use it. mv_skill_md copies it into .agents/skills/fossick/ at the root of the current git repository, which repo_root finds. It also copies it into .claude/skills/fossick/ under the working directory. With dry_run=True, the default, it prints where the file would go and copies nothing. The fossick install command copies the file.

---

source

mv_skill_md

def mv_skill_md(
    dry_run:bool=True, dir:NoneType=None
)->None:

Copy fossick’s bundled SKILL.md into the agent skill folders, or only print where it would go when dry_run is True.

---

source

repo_root

def repo_root()->pathlib.Path:

Return the nearest folder at or above the working directory that contains .git, or None.

root = repo_root()
assert root is not None and (root/'.git').exists(), f"Expected git root, got {root}"
mv_skill_md(dry_run=True)

for _bad in ('http://169.254.169.254/latest/meta-data/',
             'http://127.0.0.1:8000/admin', 'http://localhost/x', 'http://10.0.0.5/x',
             'http://192.168.1.1/', 'http://[::1]/', 'http://metadata.google.internal/',
             'http://0.0.0.0/', 'file:///etc/passwd'):
    try: check_url(_bad); raise SystemExit(f'not blocked: {_bad}')
    except BlockedURL: pass

assert check_url('https://example.com/a') == 'https://example.com/a'
assert check_url('http://127.0.0.1:8000/x', allow_private=True)

# a BlockedURL message names the override, or callers remove the address check
try: check_url('http://localhost/x')
except BlockedURL as e: assert 'allow_private' in str(e) and 'FOSSICK_ALLOW_PRIVATE' in str(e)

os.environ['FOSSICK_ALLOW_PRIVATE'] = '1'
try: assert check_url('http://127.0.0.1/x')
finally: del os.environ['FOSSICK_ALLOW_PRIVATE']