LEARN. PRACTICE. GROW.1. Define the contract before automation
An APS team prepares a rule review before migrating a reporting service. Its script must collect inventory without changing production. Define the resource, version, administrative domain, permissions and completeness criteria before writing the request loop. FMC 10.0 authentication uses token generation and specific headers; this exercise’s synthetic Bearer scheme is not FMC integration code. A dedicated account reduces interference with UI sessions. Credentials, tokens and responses containing sensitive data do not belong in the change ticket. Retain correlation identifiers, time and status, keeping secrets in an appropriate mechanism. Technical write permission also does not replace change authorization.
2. Reconcile every page
The lab returns five objects in pages of two, two and one. Stopping at the first 200 response produces an incomplete export that appears successful. The client follows approved continuation, retains IDs, rejects duplicates and cycles, and compares the advertised total when present. If the second page fails, it does not present the first as complete inventory. Even matching counts do not prove a transactional snapshot from an API allowing changes during reads. In production, document endpoint consistency and collection time span. A next reference to another origin is rejected before requesting it; credentials are not automatically sent to a URL merely because it appeared in a response.
3. Distinguish errors and bound retries
In the fictional server, 401 means the valid credential is missing and 403 identifies a role lacking resource access. Repeating the same forbidden request does not repair permission. A 429 calls for respecting limits, but the limit dimension depends on the service. The example actually waits one second and retries GET only once. The client accepts only integer Retry-After between zero and two seconds; HTTP also allows a date, which this implementation does not parse. These values are local limits rather than universal Cisco recommendations. The FMC guide documents rate and concurrency limits; identify the exhausted dimension before increasing parallelism. Invalid JSON inside a 200 response is a parsing error rather than empty inventory.
4. Resolve uncertain effects and conflicts
POST /changes creates an in-memory object and returns 503. The client records the error without automatic retry; a later read finds exactly one effect. Retrying a non-idempotent operation without knowing its outcome can duplicate it. Use correlation or idempotency mechanisms when the contract provides them and reconcile before deciding on another write. On a different resource, GET returns ETag 1; writing with If-Match 0 receives 412 and leaves state unchanged. Read the current revision, compare intent against the concurrent change and renew approval where necessary. The following experiment uses the correct version and confirms the result by reading. This demonstrates generic HTTP without asserting ETag support on a particular FMC endpoint.
5. Run and hand over scoped evidence
Run python3 run.py on a machine with Python 3.13 and permission for loopback sockets. The program opens a temporary local port, makes 24 requests, checks 30 conditions and shuts down the server and thread. It uses no external destinations, TLS, real accounts or device configuration files. Changes disappear with the process. For RUN, supplement real automation with timeouts, body limits, reconciled inventory, secret-free logs, ownership and recovery. Read-back confirms a resource representation but proves neither firewall deployment nor working service. Summary: treat authentication, completeness, retry and effect as separate problems and report exactly which stage was demonstrated.
#!/usr/bin/env python3
"""Original generic HTTP API lab. All state and credentials are synthetic."""
import hashlib,http.client,json,platform,threading,time
from collections import Counter
from http.server import BaseHTTPRequestHandler,HTTPServer
from pathlib import Path
from urllib.parse import urlsplit,urljoin,parse_qs
TOKEN='synthetic-read'
state={'requests':[],'rate':0,'changes':[],'version':1,'config':{'enabled':False}}
class Handler(BaseHTTPRequestHandler):
def log_message(self,*args):pass
def reply(self,status,body,headers=None):
raw=body if isinstance(body,bytes)else json.dumps(body).encode
self.send_response(status);self.send_header('Content-Type','application/json');self.send_header('Content-Length',str(len(raw)))
for k,v in (headers or {}).items:self.send_header(k,v)
self.end_headers;self.wfile.write(raw)
def request(self):
route=urlsplit(self.path);p=route.path;state['requests'].append({'method':self.command,'path':self.path})
if self.headers.get('Authorization')!='Bearer '+TOKEN:return self.reply(401,{'error':'synthetic credential required'},{'WWW-Authenticate':'Bearer realm="synthetic-lab"'})
if p=='/forbidden':return self.reply(403,{'error':'synthetic role lacks permission'})
if p in ['/rate','/permanent-rate']:
if p=='/rate':state['rate']+=1
if p=='/permanent-rate'or state['rate']==1:return self.reply(429,{'error':'limit'}, {'Retry-After':'1'})
return self.reply(200,{'ok':True})
if p=='/redirect':return self.reply(302,{}, {'Location':'http://example.invalid/collect'})
if p=='/invalid-json':return self.reply(200,b'{broken')
if p=='/cycle':return self.reply(200,{'items':[{'id':'r1'}],'next':'/cycle'})
if p=='/foreign':return self.reply(200,{'items':[{'id':'r1'}],'next':'http://example.invalid/collect'})
if p=='/count-mismatch':return self.reply(200,{'items':[{'id':'r1'}],'next':None,'count':5})
if p=='/duplicate':return self.reply(200,{'items':[{'id':'r1'},{'id':'r1'}],'next':None})
if p=='/broken-page':return self.reply(500,{'error':'page unavailable'})
if p=='/partial':return self.reply(200,{'items':[{'id':'r1'}],'next':'/broken-page'})
if p=='/inventory':
offset=int(parse_qs(route.query).get('offset',['0'])[0]);items=[{'id':f'r{i}','name':f'Rule {i}'}for i in range(1,6)]
return self.reply(200,{'items':items[offset:offset+2],'next':f'/inventory?offset={offset+2}'if offset+2<len(items)else None,'count':5})
if p=='/changes':
if self.command=='POST':
self.rfile.read(int(self.headers.get('Content-Length','0')));state['changes'].append({'id':'change-1'});return self.reply(503,{'error':'synthetic response error after effect'})
return self.reply(200,{'items':state['changes']})
if p=='/config':
if self.command=='PUT':
raw=self.rfile.read(int(self.headers.get('Content-Length','0')))
if self.headers.get('If-Match')!=f'"{state["version"]}"':return self.reply(412,{'error':'stale version'})
state['config']=json.loads(raw);state['version']+=1
return self.reply(200,state['config'],{'ETag':f'"{state["version"]}"'})
return self.reply(404,{'error':'unknown local resource'})
do_GET=request;do_POST=request;do_PUT=request
class APIError(Exception):pass
class Client:
def __init__(self,origin,token=TOKEN):self.origin=origin;self.token=token;self.waits=[]
def path(self,target):
dest=urlsplit(urljoin(self.origin,target));base=urlsplit(self.origin)
if (dest.scheme,dest.hostname,dest.port)!=(base.scheme,base.hostname,base.port)or dest.username or dest.password:raise APIError('unapproved-origin')
if dest.fragment:raise APIError('unexpected-fragment')
return dest.path+('?' + dest.query if dest.query else '')
def call(self,method,target,body=None,headers=None):
path=self.path(target);base=urlsplit(self.origin)
for attempt in range(2):
conn=http.client.HTTPConnection(base.hostname,base.port,timeout=2)
try:
h={'Authorization':'Bearer '+self.token,'Content-Type':'application/json',**(headers or {})};raw=None if body is None else json.dumps(body).encode
conn.request(method,path,body=raw,headers=h);response=conn.getresponse;status=response.status;rh={k.lower:v for k,v in response.getheaders};payload=response.read(16385)
finally:conn.close
if len(payload)>16384:raise APIError('body-too-large')
if status==429 and method=='GET'and attempt==0:
wait=rh.get('retry-after','')
if not wait.isdigitor not 0<=int(wait)<=2:raise APIError('retry-outside-local-budget')
self.waits.append(int(wait));time.sleep(int(wait));continue
if status<200 or status>=300:raise APIError('http-'+str(status))
try:return json.loads(payload),rh
except json.JSONDecodeError:raise APIError('invalid-json')
def inventory(self,target='/inventory?offset=0'):
seen=set;ids=set;result=[];expected=None
while target:
path=self.path(target)
if path in seen:raise APIError('pagination-cycle')
if len(seen)>=5:raise APIError('page-budget')
seen.add(path);data,_=self.call('GET',target)
if 'count'in data:
if not isinstance(data['count'],int)or data['count']<0:raise APIError('invalid-count')
if expected is not None and expected!=data['count']:raise APIError('count-drift')
expected=data['count']
if not isinstance(data.get('items'),list):raise APIError('invalid-items')
for item in data['items']:
if not isinstance(item,dict)or not isinstance(item.get('id'),str):raise APIError('invalid-item')
if item['id']in ids:raise APIError('duplicate-id')
ids.add(item['id']);result.append(item)
target=data.get('next')
if expected is not None and len(result)!=expected:raise APIError('count-mismatch')
return result
def main:
checks=[];results={}
def check(label,ok):
if not ok:raise AssertionError(label)
checks.append(label)
def error(label,fn,expected):
try:fn
except APIError as e:check(label,str(e)==expected)
else:raise AssertionError(label+' accepted')
server=HTTPServer(('127.0.0.1',0),Handler);thread=threading.Thread(target=server.serve_forever,kwargs={'poll_interval':0.05});thread.start;origin='http://127.0.0.1:'+str(server.server_port);client=Client(origin)
try:
error('missing valid token is 401',lambda:Client(origin,'invalid').call('GET','/inventory'),'http-401')
error('forbidden role is 403 without retry',lambda:client.call('GET','/forbidden'),'http-403')
inventory=client.inventory;check('five objects recovered across three pages',[x['id']for x in inventory]==['r1','r2','r3','r4','r5'])
check('names retained with IDs',inventory[-1]['name']=='Rule 5')
error('cyclic pagination rejected',lambda:client.inventory('/cycle'),'pagination-cycle')
error('foreign next link rejected before request',lambda:client.inventory('/foreign'),'unapproved-origin')
error('automatic redirect not followed',lambda:client.call('GET','/redirect'),'http-302')
error('declared count mismatch rejected',lambda:client.inventory('/count-mismatch'),'count-mismatch')
error('duplicate inventory ID rejected',lambda:client.inventory('/duplicate'),'duplicate-id')
error('incomplete inventory fails instead of returning partial success',lambda:client.inventory('/partial'),'http-500')
error('invalid JSON is not treated as empty inventory',lambda:client.call('GET','/invalid-json'),'invalid-json')
start=time.monotonic;data,_=client.call('GET','/rate');check('GET retry succeeds after one rate limit',data=={'ok':True});check('Retry-After one second actually waited',time.monotonic-start>=1)
error('persistent rate limit exhausts two-attempt budget',lambda:client.call('GET','/permanent-rate'),'http-429');check('bounded waits recorded',client.waits==[1,1])
error('POST response error does not trigger automatic retry',lambda:client.call('POST','/changes',{'name':'synthetic change'}),'http-503')
data,_=client.call('GET','/changes');check('effect exists despite POST 503',data=={'items':[{'id':'change-1'}]})
before,h=client.call('GET','/config');check('initial representation has version one',h['etag']=='"1"'and before=={'enabled':False})
error('stale If-Match rejected',lambda:client.call('PUT','/config',{'enabled':True},{'If-Match':'"0"'}),'http-412')
data,_=client.call('GET','/config');check('stale write leaves representation unchanged',data==before)
data,h=client.call('PUT','/config',{'enabled':True},{'If-Match':'"1"'});check('current If-Match applies once',data=={'enabled':True}and h['etag']=='"2"')
data,h=client.call('GET','/config');check('read-back confirms resulting representation',data=={'enabled':True}and h['etag']=='"2"')
counts=Counter((x['method'],urlsplit(x['path']).path)for x in state['requests'])
check('forbidden request sent only once',counts['GET','/forbidden']==1)
check('cycle detected after one page',counts['GET','/cycle']==1)
check('rate endpoint called twice',counts['GET','/rate']==2)
check('permanent rate endpoint bounded to two calls',counts['GET','/permanent-rate']==2)
check('side-effecting POST sent exactly once',counts['POST','/changes']==1)
check('all observed requests stay on local paths',all(x['path'].startswith('/')for x in state['requests']))
check('request evidence excludes authentication tokens',TOKEN not in json.dumps(state['requests']))
results={'inventoryIDs':[x['id']for x in inventory],'waitsSeconds':client.waits,'createdChanges':len(state['changes']),'finalConfig':state['config'],'finalVersion':state['version'],'requests':state['requests'],'requestsCount':len(state['requests'])}
finally:server.shutdown;server.server_close;thread.join(3)
check('owned HTTP server and thread stopped',not thread.is_alive)
print(json.dumps({'passed':len(checks),'failed':0,'checks':checks,'observations':results,'python':platform.python_version,'scriptSHA256':hashlib.sha256(Path(__file__).read_bytes).hexdigest,'actualLocalHTTP':True,'actualCiscoAPI':False,'actualDLP':False,'tlsUsedInThisLab':False,'externalRequests':False,'realCredentialsUsed':False,'durableWrites':False,'fmcSchemaCompatibilityClaimed':False,'scope':'Generic synthetic API; integer Retry-After only, local retry/body/page budgets, in-memory effects. No real deployment or DLP enforcement.'},indent=2))
if __name__=='__main__':main
A POST responds 503 after creating change-1. Reading finds one object; blindly repeating risks creating another.
Common pitfalls
First page as total; credentials in a redirect; indiscriminate retries; 200 as business proof; generic ETag as an FMC capability.
Related topics: Automation and APIs · Change and rollback · Observability
An operational script needs a contract, bounds, reconciliation and evidence of effect.
Reference: Secure Firewall Management Center REST API Quick Start Guide · 350-701 SCOR v2.0, effective 2026-08-27; core component of CCNP Security