dr.ptAPRENDER. PRATICAR. EVOLUIR.
← CCNP Security: núcleo SCOR e operação
17 / 25 · 55 MIN

APIs: inventário e mudança controlada

Interpreta um cliente HTTP, deteta inventários incompletos e recupera de alterações com resultado incerto.

1. Definir o contrato antes de automatizar

Uma equipa APS prepara a revisão de regras antes de migrar um serviço de relatórios. O script deve recolher o inventário sem alterar produção. Define recurso, versão, domínio administrativo, permissões e critérios de completude antes de escrever o ciclo de pedidos. No FMC 10.0, a autenticação usa geração de tokens e cabeçalhos específicos; o Bearer sintético deste exercício não é código de integração FMC. Uma conta dedicada reduz interferências com sessões da interface. Credenciais, tokens e respostas com dados sensíveis não pertencem ao ticket de mudança. Guarda identificadores de correlação, hora e estado, mantendo segredos num mecanismo apropriado. Uma permissão técnica de escrita também não substitui a autorização da mudança.

2. Reconciliar todas as páginas

O laboratório devolve cinco objetos em páginas de dois, dois e um. Parar na primeira resposta 200 produz uma exportação incompleta com aparência de sucesso. O cliente segue a continuação aprovada, conserva IDs, rejeita duplicados e ciclos e compara o total anunciado quando existe. Se a segunda página falhar, não apresenta a primeira como inventário completo. Mesmo uma contagem coincidente não prova um snapshot transacional de uma API que permite alterações durante a leitura. Em produção, documenta a consistência oferecida pelo endpoint e a janela de recolha. Uma referência next para outra origem é recusada antes do pedido; não se enviam credenciais automaticamente para um URL apenas porque apareceu numa resposta.

3. Distinguir erros e limitar repetições

No servidor fictício, 401 indica que falta a credencial válida e 403 identifica uma função sem acesso ao recurso. Repetir o mesmo pedido proibido não resolve a permissão. Um 429 pede respeito pelos limites, mas a dimensão do limite depende do serviço. O exemplo espera realmente um segundo e repete GET apenas uma vez. O cliente aceita só Retry-After inteiro entre zero e dois segundos; HTTP também permite uma data, que esta implementação não interpreta. Esses valores são limites locais, não recomendações universais Cisco. O guia FMC documenta limites de taxa e concorrência; identifica qual foi atingido antes de aumentar paralelismo. Um corpo JSON inválido numa resposta 200 é erro de interpretação, não inventário vazio.

4. Resolver efeitos incertos e conflitos

POST /changes cria um objeto em memória e devolve 503. O cliente regista o erro sem repetir automaticamente; uma leitura posterior encontra exatamente um efeito. Repetir uma operação não idempotente sem saber o resultado pode duplicá-lo. Usa correlação ou mecanismos de idempotência quando o contrato os fornece e reconcilia antes de decidir nova escrita. Noutro recurso, GET devolve ETag 1; uma escrita com If-Match 0 recebe 412 e não altera o estado. Lê a revisão atual, compara a intenção com a alteração concorrente e renova aprovação quando necessário. O ensaio seguinte usa a versão correta e confirma o resultado por leitura. Isto demonstra HTTP genérico, sem afirmar suporte ETag num endpoint FMC concreto.

5. Executar e entregar evidência limitada

Executa python3 run.py numa máquina com Python 3.13 e permissão para sockets loopback. O programa abre uma porta temporária local, faz 24 pedidos, verifica 30 condições e encerra o servidor e a thread. Não utiliza destinos externos, TLS, contas reais ou ficheiros de configuração de equipamentos. As alterações desaparecem com o processo. Para RUN, acrescenta à automação real timeouts, limites de corpo, inventário reconciliado, logs sem segredos, responsável e recuperação. A leitura após escrita confirma a representação do recurso, mas não prova deployment no firewall nem serviço funcional. Resumo: trata autenticação, completude, repetição e efeito como problemas distintos e reporta exatamente a etapa demonstrada.

#!/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.isdigit()or 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()
NA PRÁTICA

Um POST responde 503 depois de criar change-1. A leitura encontra um objeto; repetir cegamente arrisca criar outro.

Armadilhas comuns

Primeira página como total; credenciais num redirect; retry indiscriminado; 200 como prova de negócio; ETag genérico como capacidade FMC.

Tópicos relacionados: Automação e APIs · Mudança e rollback · Observabilidade

Leva esta ideia contigo

Um script operacional precisa de contrato, limites, reconciliação e evidência do efeito.

Criar conta
AnteriorPróxima aula

Referência: Secure Firewall Management Center REST API Quick Start Guide · 350-701 SCOR v2.0, effective 2026-08-27; core component of CCNP Security

CCNP® e Cisco® são marcas registadas da Cisco Systems, Inc. e/ou das suas afiliadas. A dr.pt é uma plataforma de preparação independente e não está afiliada, associada, patrocinada, autorizada nem aprovada por Cisco. Os conteúdos e as perguntas são originais, não são perguntas oficiais de exame, e concluir os nossos testes não atribui nem garante qualquer certificação. Os nomes são usados apenas para identificar o tema. Todas as outras marcas pertencem aos respetivos titulares.