Web Services, SOAP, Authentication, Tokens, and Session Workflows: Guided Hands-On Workflow

1. Safety envelope

2. Start the disposable service

Save as fixtures/auth_soap_fixture.py:

from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from http.cookies import SimpleCookie
from urllib.parse import urlparse
from pathlib import Path
from xml.etree import ElementTree as ET
import argparse
import hashlib
import json
import threading
import time

FIXTURE_VERSION = "prompt15-auth-soap-fixture-v1"
SOAP_NS = "http://schemas.xmlsoap.org/soap/envelope/"
APP_NS = "urn:devops-academy:auth"

FAKE_CREDENTIALS = {
    "user01": "fake-pass-01",
    "user02": "fake-pass-02",
    "user03": "fake-pass-03",
}

lock = threading.Lock()
session_seq = 0
sessions = {}
event_log = None
metrics = {
    "requests": 0,
    "errors": 0,
    "logins": 0,
    "refreshes": 0,
    "logouts": 0,
    "auth_failures": 0,
    "by_operation": {},
}

def now_ms():
    return int(time.time() * 1000)

def token_fp(value):
    if not value:
        return ""
    return hashlib.sha256(value.encode("utf-8")).hexdigest()[:12]

def log_event(event):
    if event_log is None:
        return
    with lock:
        with event_log.open("a", encoding="utf-8") as handle:
            handle.write(json.dumps(event, sort_keys=True) + "\n")

def metric(operation, status):
    with lock:
        metrics["requests"] += 1
        metrics["by_operation"][operation] = metrics["by_operation"].get(operation, 0) + 1
        if status >= 400:
            metrics["errors"] += 1

def bearer(header):
    if not header or not header.startswith("Bearer "):
        return ""
    return header[len("Bearer "):].strip()

def session_cookie(header):
    if not header:
        return ""
    jar = SimpleCookie()
    try:
        jar.load(header)
    except Exception:
        return ""
    morsel = jar.get("LABSESSION")
    return morsel.value if morsel else ""

def soap_envelope(body_xml):
    return f'''<?xml version="1.0" encoding="UTF-8"?>
<soap:Envelope xmlns:soap="{SOAP_NS}" xmlns:m="{APP_NS}">
  <soap:Body>
    {body_xml}
  </soap:Body>
</soap:Envelope>'''.encode("utf-8")

class Handler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def read_body(self, limit=65536):
        length = int(self.headers.get("Content-Length", "0") or "0")
        if length > limit:
            raise ValueError("body too large")
        return self.rfile.read(length) if length else b""

    def read_json(self):
        raw = self.read_body()
        return json.loads(raw.decode("utf-8") if raw else "{}")

    def send_json(self, status, payload, extra_headers=None):
        raw = json.dumps(payload, sort_keys=True).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(raw)))
        self.send_header("X-Fixture-Version", FIXTURE_VERSION)
        for k, v in (extra_headers or {}).items():
            self.send_header(k, v)
        self.end_headers()
        self.wfile.write(raw)

    def send_xml(self, status, raw, extra_headers=None):
        self.send_response(status)
        self.send_header("Content-Type", "text/xml; charset=utf-8")
        self.send_header("Content-Length", str(len(raw)))
        self.send_header("X-Fixture-Version", FIXTURE_VERSION)
        for k, v in (extra_headers or {}).items():
            self.send_header(k, v)
        self.end_headers()
        self.wfile.write(raw)

    def send_empty(self, status, extra_headers=None):
        self.send_response(status)
        self.send_header("Content-Length", "0")
        self.send_header("X-Fixture-Version", FIXTURE_VERSION)
        for k, v in (extra_headers or {}).items():
            self.send_header(k, v)
        self.end_headers()

    def record_event(self, started, operation, status, username="", sid="", token="", auth_state="", extra=None):
        metric(operation, status)
        event = {
            "ts_ms": now_ms(),
            "operation": operation,
            "method": self.command,
            "path": urlparse(self.path).path,
            "status": status,
            "service_wall_ms": now_ms() - started,
            "username": username,
            "session_id": sid,
            "token_fp": token_fp(token),
            "authorization": "Bearer <redacted>" if self.headers.get("Authorization") else "",
            "auth_state": auth_state,
        }
        if extra:
            event.update(extra)
        log_event(event)

    def authorize_access(self):
        sid = session_cookie(self.headers.get("Cookie"))
        token = bearer(self.headers.get("Authorization"))
        if not sid or not token:
            return False, 401, "missing_auth", "", sid, token

        with lock:
            session = sessions.get(sid)
            if not session or not session["active"]:
                return False, 401, "invalid_session", "", sid, token
            if token != session["access_token"]:
                return False, 401, "invalid_token", session["username"], sid, token
            if session["access_uses_left"] <= 0:
                metrics["auth_failures"] += 1
                return False, 401, "token_expired", session["username"], sid, token
            session["access_uses_left"] -= 1
            username = session["username"]
            remaining = session["access_uses_left"]
        return True, 200, "authorized", username, sid, token, remaining

    def do_GET(self):
        started = now_ms()
        path = urlparse(self.path).path

        if path == "/health":
            status = 200
            self.send_json(status, {"status": "ok", "fixture_version": FIXTURE_VERSION})
            self.record_event(started, "health", status)
            return

        if path == "/stats":
            with lock:
                active = {
                    sid: {
                        "username": s["username"],
                        "access_uses_left": s["access_uses_left"],
                        "generation": s["generation"],
                    }
                    for sid, s in sessions.items()
                    if s["active"]
                }
                snapshot = {
                    "fixture_version": FIXTURE_VERSION,
                    "active_sessions": len(active),
                    "sessions": active,
                    "metrics": dict(metrics),
                }
            status = 200
            self.send_json(status, snapshot)
            self.record_event(started, "stats", status)
            return

        if path == "/api/profile":
            auth = self.authorize_access()
            if not auth[0]:
                _, status, reason, username, sid, token = auth
                self.send_json(status, {"status": reason})
                self.record_event(started, "profile", status, username, sid, token, reason)
                return
            _, _, _, username, sid, token, remaining = auth
            time.sleep(0.015)
            status = 200
            self.send_json(status, {
                "status": "ok",
                "username": username,
                "session_id": sid,
                "access_uses_left": remaining,
            })
            self.record_event(started, "profile", status, username, sid, token, "authorized")
            return

        status = 404
        self.send_json(status, {"status": "not_found", "path": path})
        self.record_event(started, "unknown_get", status)

    def do_POST(self):
        global session_seq
        started = now_ms()
        path = urlparse(self.path).path

        if path == "/soap/ping":
            try:
                root = ET.fromstring(self.read_body())
                message = root.findtext(f".//{{{APP_NS}}}Message") or ""
            except Exception:
                status = 400
                self.send_xml(status, soap_envelope(
                    '<soap:Fault><faultcode>soap:Client</faultcode><faultstring>Invalid XML</faultstring></soap:Fault>'
                ))
                self.record_event(started, "soap_ping", status, auth_state="invalid_xml")
                return

            status = 200
            raw = soap_envelope(
                f'<m:PingResponse><m:Pong>{message}</m:Pong><m:Fixture>{FIXTURE_VERSION}</m:Fixture></m:PingResponse>'
            )
            self.send_xml(status, raw)
            self.record_event(started, "soap_ping", status, auth_state="public")
            return

        if path == "/auth/login":
            try:
                body = self.read_json()
            except Exception:
                status = 400
                self.send_json(status, {"status": "invalid_json"})
                self.record_event(started, "login", status, auth_state="invalid_json")
                return

            username = str(body.get("username", ""))
            password = str(body.get("password", ""))
            if FAKE_CREDENTIALS.get(username) != password:
                with lock:
                    metrics["auth_failures"] += 1
                status = 401
                self.send_json(status, {"status": "invalid_credentials"})
                self.record_event(started, "login", status, username=username, auth_state="invalid_credentials")
                return

            with lock:
                session_seq += 1
                sid = f"SID-{session_seq:06d}"
                generation = 1
                access = f"AT-{session_seq:06d}-G{generation}"
                refresh = f"RT-{session_seq:06d}-G{generation}"
                sessions[sid] = {
                    "username": username,
                    "generation": generation,
                    "access_token": access,
                    "refresh_token": refresh,
                    "access_uses_left": 2,
                    "active": True,
                }
                metrics["logins"] += 1

            time.sleep(0.020)
            status = 200
            self.send_json(status, {
                "status": "logged_in",
                "access_token": access,
                "refresh_token": refresh,
                "session_id": sid,
                "access_uses": 2,
            }, {
                "Set-Cookie": f"LABSESSION={sid}; Path=/; HttpOnly; SameSite=Lax"
            })
            self.record_event(started, "login", status, username, sid, access, "logged_in")
            return

        if path == "/auth/refresh":
            sid = session_cookie(self.headers.get("Cookie"))
            try:
                body = self.read_json()
            except Exception:
                status = 400
                self.send_json(status, {"status": "invalid_json"})
                self.record_event(started, "refresh", status, sid=sid, auth_state="invalid_json")
                return
            supplied_refresh = str(body.get("refresh_token", ""))

            with lock:
                session = sessions.get(sid)
                if not session or not session["active"]:
                    ok = False
                    reason = "invalid_session"
                    username = ""
                elif supplied_refresh != session["refresh_token"]:
                    ok = False
                    reason = "invalid_refresh"
                    username = session["username"]
                else:
                    ok = True
                    username = session["username"]
                    session["generation"] += 1
                    generation = session["generation"]
                    suffix = sid.split("-")[-1]
                    access = f"AT-{suffix}-G{generation}"
                    refresh = f"RT-{suffix}-G{generation}"
                    session["access_token"] = access
                    session["refresh_token"] = refresh
                    session["access_uses_left"] = 2
                    metrics["refreshes"] += 1

            if not ok:
                with lock:
                    metrics["auth_failures"] += 1
                status = 401
                self.send_json(status, {"status": reason})
                self.record_event(started, "refresh", status, username, sid, supplied_refresh, reason)
                return

            time.sleep(0.018)
            status = 200
            self.send_json(status, {
                "status": "refreshed",
                "access_token": access,
                "refresh_token": refresh,
                "session_id": sid,
                "access_uses": 2,
            })
            self.record_event(started, "refresh", status, username, sid, access, "refreshed")
            return

        if path == "/auth/logout":
            sid = session_cookie(self.headers.get("Cookie"))
            try:
                body = self.read_json()
            except Exception:
                body = {}
            supplied_refresh = str(body.get("refresh_token", ""))

            with lock:
                session = sessions.get(sid)
                if not session or not session["active"]:
                    ok = False
                    username = ""
                elif supplied_refresh != session["refresh_token"]:
                    ok = False
                    username = session["username"]
                else:
                    ok = True
                    username = session["username"]
                    session["active"] = False
                    metrics["logouts"] += 1

            if not ok:
                status = 401
                self.send_json(status, {"status": "logout_rejected"})
                self.record_event(started, "logout", status, username, sid, supplied_refresh, "logout_rejected")
                return

            status = 204
            self.send_empty(status, {
                "Set-Cookie": "LABSESSION=; Path=/; Max-Age=0; HttpOnly; SameSite=Lax"
            })
            self.record_event(started, "logout", status, username, sid, supplied_refresh, "logged_out")
            return

        if path == "/soap/account":
            auth = self.authorize_access()
            if not auth[0]:
                _, status, reason, username, sid, token = auth
                fault = soap_envelope(
                    f'<soap:Fault><faultcode>soap:Client.Auth</faultcode><faultstring>{reason}</faultstring></soap:Fault>'
                )
                self.send_xml(status, fault)
                self.record_event(started, "soap_account", status, username, sid, token, reason)
                return

            _, _, _, username, sid, token, remaining = auth
            try:
                root = ET.fromstring(self.read_body())
                requested_sid = root.findtext(f".//{{{APP_NS}}}SessionId") or ""
            except Exception:
                status = 400
                self.send_xml(status, soap_envelope(
                    '<soap:Fault><faultcode>soap:Client</faultcode><faultstring>Invalid XML</faultstring></soap:Fault>'
                ))
                self.record_event(started, "soap_account", status, username, sid, token, "invalid_xml")
                return

            if requested_sid != sid:
                status = 409
                self.send_xml(status, soap_envelope(
                    '<soap:Fault><faultcode>soap:Client.Session</faultcode><faultstring>Session mismatch</faultstring></soap:Fault>'
                ))
                self.record_event(started, "soap_account", status, username, sid, token, "session_mismatch",
                            {"requested_session_id": requested_sid})
                return

            time.sleep(0.024)
            status = 200
            raw = soap_envelope(
                f'''<m:AccountSummaryResponse>
  <m:User>{username}</m:User>
  <m:SessionId>{sid}</m:SessionId>
  <m:AuthState>valid</m:AuthState>
  <m:AccessUsesLeft>{remaining}</m:AccessUsesLeft>
</m:AccountSummaryResponse>'''
            )
            self.send_xml(status, raw)
            self.record_event(started, "soap_account", status, username, sid, token, "authorized")
            return

        status = 404
        self.send_json(status, {"status": "not_found", "path": path})
        self.record_event(started, "unknown_post", status)

    def log_message(self, format, *args):
        return

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=8000)
    parser.add_argument("--log", default="results/auth-events.jsonl")
    args = parser.parse_args()

    global event_log
    event_log = Path(args.log).resolve()
    event_log.parent.mkdir(parents=True, exist_ok=True)
    event_log.write_text("", encoding="utf-8")

    server = ThreadingHTTPServer((args.host, args.port), Handler)
    print(f"fixture_version={FIXTURE_VERSION}")
    print(f"base_url=http://{args.host}:{args.port}")
    print(f"event_log={event_log}")
    server.serve_forever()

if __name__ == "__main__":
    main()

Start:

python fixtures/auth_soap_fixture.py --host 127.0.0.1 --port 8000 --log results/auth-events.jsonl

3. Read-only preflight

curl --fail --silent http://127.0.0.1:8000/health
curl --fail --silent http://127.0.0.1:8000/stats

Expected: fixture version prompt15-auth-soap-fixture-v1, zero active sessions, zero logins/refreshes/logouts.

4. Create a fake credential source

data/fake-credentials.csv:

username,password
user01,fake-pass-01
user02,fake-pass-02

CSV Data Set Config:

With 2 threads ×1 journey, two different fake rows are consumed; thread-to-row order is scheduler-dependent.

5. Prove SOAP/XML before authentication

HTTP Request — SOAP Ping:

POST /soap/ping
Content-Type: text/xml; charset=utf-8
SOAPAction: "urn:Ping"

<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/"
               xmlns:m="urn:devops-academy:auth">
  <soap:Body>
    <m:Ping><m:Message>Hello JMeter</m:Message></m:Ping>
  </soap:Body>
</soap:Envelope>

Add XML Assertion. Add XPath2 Assertion aliases soap/m and assert Pong=Hello JMeter. Add XPath2 Extractor PONG for the Pong text.

6. Add HTTP Cookie Manager

Add one Cookie Manager at Thread Group scope. Do not manually define LABSESSION. Login's Set-Cookie response populates the current thread's jar automatically.

7. Login and extract per-thread identity state

POST /auth/login
Content-Type: application/json
Accept: application/json

{
  "username": "${username}",
  "password": "${password}"
}

Assert HTTP 200, status=logged_in, access_uses=2. Add JSON JMESPath Extractors:

8. Protected JSON call — access use #1

GET /api/profile
Authorization: Bearer ${ACCESS_TOKEN}
Accept: application/json

Scope this Header Manager only to protected samplers. Assert HTTP 200, status=ok, correct username/session, and access_uses_left=1.

9. Authenticated SOAP call — access use #2

POST /soap/account
Authorization: Bearer ${ACCESS_TOKEN}
Content-Type: text/xml; charset=utf-8
SOAPAction: "urn:GetAccountSummary"

<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/"
               xmlns:m="urn:devops-academy:auth">
  <soap:Body>
    <m:GetAccountSummary>
      <m:SessionId>${SESSION_ID}</m:SessionId>
    </m:GetAccountSummary>
  </soap:Body>
</soap:Envelope>

Add XML Assertion plus XPath2 Assertion with aliases:

soap=http://schemas.xmlsoap.org/soap/envelope/
m=urn:devops-academy:auth

boolean(
  /soap:Envelope/soap:Body/m:AccountSummaryResponse[
    m:User='${username}' and
    m:SessionId='${SESSION_ID}' and
    m:AuthState='valid' and
    m:AccessUsesLeft='0'
  ]
)

Add XPath2 Extractor SOAP_SESSION_ID and assert it matches ${SESSION_ID}.

10. Trigger and preserve expiry

Call Expired Token Probe with the same Authorization header. The target returns HTTP 401 and JSON {"status":"token_expired"}. Leave the sample failed and assert the JSON reason. Set Thread Group error action to Continue only for this bounded workflow so Refresh and Logout still execute.

Do not use Ignore Status or a script to mark this expected auth failure green.

11. Refresh rotates tokens

POST /auth/refresh
Content-Type: application/json

{
  "refresh_token": "${REFRESH_TOKEN}"
}

Cookie Manager supplies LABSESSION. Assert 200/refreshed/session_id=${SESSION_ID}. Reuse the extractor variable names ACCESS_TOKEN and REFRESH_TOKEN so generation 2 overwrites generation 1.

12. Verify post-refresh protected access

Run Profile After Refresh with Authorization: Bearer ${ACCESS_TOKEN}. JMeter resolves the variable at execution time, so the new token is used. Assert correct user/session and HTTP 200.

13. Logout invalidates session state

POST /auth/logout
Content-Type: application/json

{
  "refresh_token": "${REFRESH_TOKEN}"
}

Cookie Manager supplies LABSESSION. Assert 204. The fixture marks the session inactive and clears the cookie.

14. Progressive JMeter tree

Thread Group — 1 or 2 threads × 1 journey
Action after Sampler error: Continue
├── HTTP Cookie Manager
├── SOAP Ping -> XML + XPath2 checks
├── Login -> extract ACCESS_TOKEN / REFRESH_TOKEN / SESSION_ID
├── Profile Before Expiry [scoped Authorization]
├── SOAP Account Before Expiry [scoped Authorization + SOAP headers]
├── Expired Token Probe [expected failed 401]
├── Refresh Token -> overwrite ACCESS_TOKEN / REFRESH_TOKEN
├── Profile After Refresh
└── Logout

15. Run in CLI with leak-resistant result settings

jmeter.bat -n `
  -t plans\auth-soap.jmx `
  -l results\auth-soap\results.jtl `
  -j results\auth-soap\jmeter.log `
  -Jjmeter.save.saveservice.requestHeaders=false `
  -Jjmeter.save.saveservice.responseHeaders=false `
  -Jjmeter.save.saveservice.samplerData=false `
  -Jjmeter.save.saveservice.response_data=false `
  -Jjmeter.save.saveservice.response_data.on_error=false `
  -Jjmeter.save.saveservice.url=false `
  -Jjmeter.save.saveservice.assertion_results_failure_message=true

Bash uses jmeter with the same options.

Preserve the raw JTL file together with its matching jmeter.log; the JTL carries timing/status/assertion evidence while the log preserves engine/runtime diagnostics without raw credentials.

16. Analyze redacted target evidence

tools/analyze_auth_events.py:

import json
import sys
from collections import Counter, defaultdict
from pathlib import Path

path = Path(sys.argv[1] if len(sys.argv) > 1 else "results/auth-events.jsonl")
events = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]
if not events:
    raise SystemExit("No auth events found")

print(f"events={len(events)}")
print(f"operations={dict(Counter(e['operation'] for e in events))}")
print(f"statuses={dict(Counter(e['status'] for e in events))}")
print(f"auth_states={dict(Counter(e.get('auth_state','') for e in events))}")

sessions = defaultdict(lambda: {"users": set(), "token_fps": set(), "operations": Counter()})
for e in events:
    sid = e.get("session_id", "")
    if sid:
        if e.get("username"):
            sessions[sid]["users"].add(e["username"])
        if e.get("token_fp"):
            sessions[sid]["token_fps"].add(e["token_fp"])
        sessions[sid]["operations"][e["operation"]] += 1

print("sessions:")
for sid, info in sorted(sessions.items()):
    print(
        f"  {sid}: users={sorted(info['users'])} "
        f"token_fps={sorted(info['token_fps'])} operations={dict(info['operations'])}"
    )

cross_user = {
    sid: sorted(info["users"])
    for sid, info in sessions.items()
    if len(info["users"]) > 1
}
print(f"cross_user_sessions={cross_user}")
print(f"expired_samples={sum(e.get('auth_state')=='token_expired' for e in events)}")
print(f"raw_authorization_values={sum(e.get('authorization') not in ('', 'Bearer <redacted>') for e in events)}")

Expected for 2 threads: two distinct session IDs, one username per session, token fingerprints rather than token strings, one token-expired event per thread, successful refreshes/logouts, and zero raw Authorization values.

17. Scan retained artifacts

tools/scan_auth_artifacts.py:

import re
import sys
from pathlib import Path

if len(sys.argv) < 2:
    raise SystemExit("usage: scan_auth_artifacts.py <file-or-directory> [...]")

patterns = {
    "access_token": re.compile(r"\bAT-\d{6}-G\d+\b"),
    "refresh_token": re.compile(r"\bRT-\d{6}-G\d+\b"),
    "fake_password": re.compile(r"\bfake-pass-\d+\b"),
    "authorization_value": re.compile(r"(?i)Authorization\s*[:=]\s*Bearer\s+(?!<redacted>)[^\s,<]+"),
}

extensions = {".jtl", ".log", ".jsonl", ".txt", ".csv", ".html"}

hits = []
for raw in sys.argv[1:]:
    path = Path(raw)
    files = [path] if path.is_file() else [
        p for p in path.rglob("*") if p.is_file() and p.suffix.lower() in extensions
    ]
    for file in files:
        try:
            text = file.read_text(encoding="utf-8", errors="replace")
        except Exception:
            continue
        for name, pattern in patterns.items():
            found = sorted(set(pattern.findall(text)))
            if found:
                hits.append((str(file), name, found[:5]))

if hits:
    for file, name, found in hits:
        print(f"HIT {file} {name}: {found}")
    raise SystemExit(
        "Retained-artifact scan failed. Investigate the source; do not weaken the scanner."
    )

print("PASS: no configured credential/token values found in retained artifacts.")

python tools/scan_auth_artifacts.py   results/auth-soap/results.jtl   results/auth-soap/jmeter.log   results/auth-events.jsonl

Do not scan the intentional fake credential source as if it were result evidence; in a real environment even credential-source files must be protected.

18. Verify cleanup/session state

curl --fail --silent http://127.0.0.1:8000/stats

After logout, active_sessions should be zero.

19. Challenge

A user journey should perform 20 protected calls after one login, but Login is inside the 20-count Loop Controller. What should move?

Move Login/extraction outside the business loop; keep cookie/token state at user scope; refresh only if the lifecycle requires it. Otherwise the test multiplies IdP load unintentionally.

Knowledge check

Official references and version notes

• Component Reference — HTTP Request, HTTP Cookie Manager, HTTP Authorization Manager, Header Manager, JSON JMESPath Extractor/Assertion, XML Assertion, XPath2 Extractor/Assertion, and result-scope semantics.
• Elements of a Test Plan — execution/scope and thread-local variable behavior.
• Getting Started — Java requirements, TLS/runtime basics, and GUI-versus-CLI guidance.
• Properties Reference — CSV result defaults, request/response header/body retention, TLS defaults, and HTTP client settings.
• Generating Dashboard Report — required result fields and dashboard interpretation.
• Best Practices — non-GUI load execution, listener restraint, and injector validity.
• Apache JMeter downloads — current stable release and Java requirement.