">
 

Fail Closed at Socket Time: A Loopback Probe for Zero-Bill Indie APIs

Iniciado por joomlamz, Hoje at 10:25

Respostas: 1   |   Visualizações: 1

Tópico anterior - Tópico seguinte

0 Membros e 1 Visitante estão a ver este tópico.

Saudações à comunidade do **webmastersmz.com**. Como especialista em tecnologia, analisei o artigo técnico sobre a implementação de um mecanismo de *"Fail Closed at Socket Time"* para APIs independentes de baixo custo. Este é um tópico crucial para qualquer administrador que dependa de serviços externos e precise de garantir a resiliência do seu ecossistema.

### Análise Técnica: O conceito de "Fail Closed" em APIs

O artigo propõe uma estratégia robusta para evitar o desperdício de recursos e custos inesperados em APIs que operam sob modelos de pagamento por uso (ou com quotas limitadas). Em vez de esperar que um *timeout* padrão do sistema ocorra — o que frequentemente leva a "cascatas de falhas" — a abordagem de **Loopback Probe** atua na camada do *socket* para encerrar conexões instantaneamente se as condições de integridade não forem atendidas.

**Pontos-chave do artigo:**

1.  **Redução de Overload:** Ao implementar um *Fail Closed* prematuro, evitamos que o nosso servidor mantenha conexões "penduradas" (*zombie sockets*) aguardando respostas de APIs que já sabemos estarem inacessíveis ou sobrecarregadas.
2.  **Proteção de Faturação:** Para APIs "Indie" ou de terceiros, cada requisição processada conta. O *Loopback Probe* atua como um gatekeeper, validando a disponibilidade do endpoint antes que a requisição real seja enviada, economizando créditos e evitando erros 4xx/5xx que podem levar ao bloqueio da nossa chave de API.
3.  **Implementação de Latência Zero:** A técnica foca na otimização da camada de rede, garantindo que o impacto no desempenho seja praticamente nulo, utilizando sondagens leves antes da execução da lógica de negócio.

**Incentivo ao Debate:**

Esta técnica é particularmente relevante para quem gere projetos aqui no nosso contexto local, onde a estabilidade das conexões internacionais nem sempre é garantida. Pergunto aos membros do fórum:
*   Como têm lidado com a gestão de timeouts em APIs de terceiros nos vossos projetos em Moçambique?
*   Alguém já implementou mecanismos semelhantes de *circuit breaking* para evitar o consumo desnecessário de quotas?

Deixo este tema aberto para trocarmos experiências sobre como otimizar a infraestrutura das nossas aplicações.

---

Para garantir que os vossos projetos e fóruns rodam sem falhas, com a estabilidade e a velocidade que os utilizadores moçambicanos exigem, convido-vos a conhecer as soluções de alojamento de alta performance da **AplicHost** em https://aplichost.com.

Fail Closed at Socket Time: A Loopback Probe for Zero-Bill Indie APIs



Tópico: Fail Closed at Socket Time: A Loopback Probe for Zero-Bill Indie APIs
Categoria: Tutoriais | Programação & Tecnologia
Idioma Principal: Português (Conteúdo de Tecnologia)

Descrição do Conteúdo / Informações:
-------------------------------------------------------------------------
A weekend API stays free only if the process cannot leave the machine. Agent-written patches still compile when they import a billed SDK and open a TCP session on the first request. A fail-closed egress probe at boot lets a solo founder ship today, keep the wallet closed, and defer every third-party host until a human edits an allowlist.

Vibe-coded backends fail as engineering when the network is implicit. The model is not the leak. The first outbound socket is.



The cost path is a hostname, not a feature flag


Indie stacks accrue charges in three boring ways. A generated client calls a hosted model. A "temporary" object store lands on a public bucket. A logging snippet posts traces to a vendor. None of those require a product decision. They require create_connection.

A committed host allowlist is smaller than a cloud policy engine. It is also dumber, which is the point. The process starts only if every resolved peer is loopback. Anything else raises before the ASGI server binds.

This is not a security product. It is a wallet latch for a one-person repo that must ship this week.



Artifact: a fail-closed connector wrap


The working unit is three files. allowlist.json is the policy. egress_guard.py installs the wrap before routers import. test_egress_guard.py proves both the allow path and the deny path. No cloud account is required to run them.



1. Commit a loopback-only policy


Keep the file tiny. Review it in the same PR as any new dependency.

{
"allowed_hosts": ["127.0.0.1", "localhost", "::1"],
"allowed_ports": [8000, 8080],
"deny_message": "egress blocked: host not on loopback allowlist"
}

Ports are optional. A stricter operator allows any port on loopback and zero ports elsewhere. The JSON exists so an agent cannot "helpfully" add api.stripe.com inside a Python literal without a file diff.



2. Wrap name lookup and TCP connect before app import


socket.create_connection is the common path for urllib, requests, and many SDKs. getaddrinfo is the earlier path. Both get wrapped. The original functions stay on the module so tests can still bind a local server.

# egress_guard.py
from __future__ import annotations

import json
import socket
from pathlib import Path
from typing import Any

POLICY_PATH = Path(__file__).with_name("allowlist.json")

class EgressBlocked(RuntimeError):
"""Raised when a peer is outside the committed allowlist."""

def load_policy() -> dict[str, Any]:
data = json.loads(POLICY_PATH.read_text(encoding="utf-8"))
hosts = frozenset(data["allowed_hosts"])
ports = frozenset(int(p) for p in data.get("allowed_ports", []))
return {"hosts": hosts, "ports": ports, "message": data["deny_message"]}

_POLICY = load_policy()
_orig_getaddrinfo = socket.getaddrinfo
_orig_create_connection = socket.create_connection

def _check(host: str, port: int | None) -> None:
if host not in _POLICY["hosts"]:
raise EgressBlocked(f"{_POLICY['message']}: host={host!r}")
if port is not None and _POLICY["ports"] and port not in _POLICY["ports"]:
raise EgressBlocked(f"{_POLICY['message']}: port={port}")

def guarded_getaddrinfo(host, port, *args, **kwargs):
_check(str(host), int(port) if port else None)
return _orig_getaddrinfo(host, port, *args, **kwargs)

def guarded_create_connection(address, *args, **kwargs):
host, port = address[0], address[1]
_check(str(host), int(port) if port is not None else None)
return _orig_create_connection(address, *args, **kwargs)

def install() -> None:
socket.getaddrinfo = guarded_getaddrinfo
socket.create_connection = guarded_create_connection

Call install() in a boot module, not in a router. Routers are what the agent edits. The boot module should be listed in a one-line code-owner rule or a CODEOWNERS file so it does not drift.

# boot.py
from egress_guard import install

install()

from app import create_app  # local import after the wrap

app = create_app()

if __name__ == "__main__":
import uvicorn

uvicorn.run(app, host="127.0.0.1", port=8000, log_level="info")

Bind 127.0.0.1, not 0.0.0.0. A loopback allowlist is worthless if the public interface is already advertised.



3. Keep the app itself on local adapters


The sample service stores a waitlist in SQLite. No Redis. No hosted queue. The handler never constructs an HTTP client. That is the product for week one: an endpoint that works on a laptop and on a single free process.

# app.py
from __future__ import annotations

import sqlite3
from pathlib import Path

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

DB_PATH = Path(__file__).with_name("waitlist.db")

class Signup(BaseModel):
email: EmailStr

def create_app() -> FastAPI:
app = FastAPI(title="local-waitlist")
conn = sqlite3.connect(DB_PATH, check_same_thread=False)
conn.execute(
"CREATE TABLE IF NOT EXISTS waitlist (email TEXT PRIMARY KEY, created_at TEXT DEFAULT CURRENT_TIMESTAMP)"
)
conn.commit()

@app.get("/health")
def health() -> dict[str, str]:
return {"status": "ok", "store": "sqlite"}

@app.post("/signup")
def signup(body: Signup) -> dict[str, str]:
conn.execute("INSERT OR IGNORE INTO waitlist(email) VALUES (?)", (body.email,))
conn.commit()
return {"email": body.email, "stored": "local"}

return app



4. Prove deny and allow with pytest


The deny test must not depend on the public internet being up. The assertion is the exception type, not a timeout. The allow test binds a tiny local server on an allowlisted port.

# test_egress_guard.py
import socket
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer

import pytest

import egress_guard
from egress_guard import EgressBlocked

@pytest.fixture(autouse=True)
def _install_guard():
egress_guard.install()
yield
socket.getaddrinfo = egress_guard._orig_getaddrinfo
socket.create_connection = egress_guard._orig_create_connection

def test_blocks_public_hostname():
with pytest.raises(EgressBlocked):
socket.create_connection(("example.com", 443), timeout=1)

def test_blocks_literal_public_ip():
with pytest.raises(EgressBlocked):
socket.create_connection(("1.1.1.1", 443), timeout=1)

def test_allows_loopback_http():
server = HTTPServer(("127.0.0.1", 8080), BaseHTTPRequestHandler)
thread = threading.Thread(target=server.handle_request, daemon=True)
thread.start()
sock = socket.create_connection(("127.0.0.1", 8080), timeout=2)
sock.close()
thread.join(timeout=2)

Run the loop the same way every time. Fail closed means CI red is the success case for a sneaky host.

python -m pip install fastapi pydantic uvicorn pytest
python -m pytest -q test_egress_guard.py
python boot.py

A second check belongs in the shell before a public bind. If allowlist.json grows a non-loopback host, the command exits non-zero.

python - <<'PY'
import json, sys
from pathlib import Path
allowed = {"127.0.0.1", "localhost", "::1"}
data = json.loads(Path("allowlist.json").read_text())
bad = [h for h in data["allowed_hosts"] if h not in allowed]
if bad:
print("non-loopback hosts:", bad)
sys.exit(1)
print("allowlist: loopback only")
PY



Decision table: ship local or open a host


Need this week
Local substitute
Open the allowlist?

Persist signups
SQLite file next to the app
No

Background work
In-process queue or a thread
No

Auth for a demo
Signed cookies, one secret in env
No

File uploads
Disk under ./data

No

Transactional email
Log the payload, send by hand
Not yet

Card capture
Do not ship the route
Not this week

Hosted inference
Defer; keep the handler offline
Only after a human review

The table is the product spec. If a generated patch needs a row that says "yes", the patch waits. Shipping today means the left column, not a vendor SDK.



Where a free model endpoint fits


Drafting the guard does not require a paid coding seat. MonkeyCode is an open-source coding assistant with free model access and a free server option, which is enough for a solo founder to generate the interceptor and run the API without attaching a card. Disclosure: This article was prepared as part of MonkeyCode's product outreach.

The probe does not depend on that tooling. It is ordinary CPython. A free server is useful only after test_egress_guard.py is green and boot.py still binds loopback. Treat the remote process like another laptop: same allowlist, same SQLite file, same refusal to add hosts.

Operators who already draft against a free model endpoint can paste egress_guard.py into that repo and run the tests before the next bind.



What this does not catch


The wrap is a latch, not a sandbox. socket.socket().connect() can still be called directly on some paths. asyncio openers, httpx with a custom transport, subprocess to curl, and HTTP_PROXY environment variables sit outside this file. IPv4-mapped IPv6 literals such as ::ffff:1.1.1.1 need an extra normalize step if the runtime accepts them. Docker --network host and Kubernetes sidecars ignore a process-level wrap.

SQLite will not survive multi-instance writes. Loopback binding will not satisfy a mobile client on another device without an explicit tunnel. Those are accepted limits for a zero-bill week, not defects to paper over with a paid queue.

Reload the policy only from disk at process start. Hot reload invites an agent to widen the list after tests pass.



Who should skip this


Skip the probe if the product must talk to a payment processor, an email provider, or a third-party webhook on day one. Skip it if the team already runs shared staging with managed identity. Skip it if the operator needs this wrap to stop data exfiltration; use an OS-level network policy or an isolated VM instead.

Solo founders who can accept a local store, a loopback bind, and a human-owned allowlist can ship the waitlist today. The wallet stays closed until someone edits JSON on purpose.


Joomlamz
Consultoria em Informática
-------------------------------------------------------
Especialista em Sistemas Web & Manutenção de Servidores.
A desenvolver o novo AplPortal com suporte a PHP 8.
Precisa de ajuda profissional? Contacte-me.

Tags: