Common Errors & Troubleshooting

requests.exceptions.SSLError: HTTPSConnectionPool(host='your-server', port=443):
Max retries exceeded with url: /piwebapi/ (Caused by SSLError(
SSLCertVerificationError(1, '[SSL: CERTIFICATE_VERIFY_FAILED]
certificate verify failed: unable to get local issuer certificate')))

# Extract the full certificate chain from the PI Web API server
openssl s_client -connect your-server:443 -showcerts < /dev/null 2>/dev/null \
  | openssl x509 -outform PEM > pi-web-api-cert.pem

# Verify the extracted certificate
openssl x509 -in pi-web-api-cert.pem -noout -subject -issuer -dates

# If there's an intermediate CA, save the full chain:
openssl s_client -connect your-server:443 -showcerts < /dev/null 2>/dev/null \
  | awk '/BEGIN CERT/,/END CERT/{print}' > pi-web-api-chain.pem

# Point your session to the extracted certificate
session.verify = "/path/to/pi-web-api-cert.pem"

# Or set via environment variable (applies to all requests in the process)
# export REQUESTS_CA_BUNDLE=/path/to/pi-web-api-cert.pem

# This makes Python's requests library trust certificates
# from the Windows certificate store automatically
pip install python-certifi-win32

# After installation, if your org's CA cert is in the Windows
# trust store, requests will automatically trust PI Web API's cert.
# No code changes needed.

# Debian/Ubuntu
sudo cp pi-web-api-cert.pem /usr/local/share/ca-certificates/pi-web-api.crt
sudo update-ca-certificates

# RHEL/CentOS/Fedora
sudo cp pi-web-api-cert.pem /etc/pki/ca-trust/source/anchors/
sudo update-ca-trust

import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

session.verify = False

"""Diagnose SSL certificate issues with PI Web API."""
import ssl
import socket

def debug_ssl(hostname, port=443):
    """Show certificate details for a PI Web API server."""
    context = ssl.create_default_context()
    context.check_hostname = False
    context.verify_mode = ssl.CERT_NONE

    with socket.create_connection((hostname, port)) as sock:
        with context.wrap_socket(sock, server_hostname=hostname) as ssock:
            cert = ssock.getpeercert(binary_form=False)
            der_cert = ssock.getpeercert(binary_form=True)

            if cert:
                print(f"Subject: {cert.get('subject')}")
                print(f"Issuer: {cert.get('issuer')}")
                print(f"Not Before: {cert.get('notBefore')}")
                print(f"Not After: {cert.get('notAfter')}")
                print(f"SANs: {cert.get('subjectAltName')}")
            else:
                # Self-signed cert with no parsed fields
                pem = ssl.DER_cert_to_PEM_cert(der_cert)
                print("Certificate (PEM):")
                print(pem[:200] + "...")

            print(f"Protocol: {ssock.version()}")
            print(f"Cipher: {ssock.cipher()}")

debug_ssl("your-pi-web-api-server")

"""Diagnose authentication issues with PI Web API."""

def debug_auth(base_url):
    """Check what authentication methods the server accepts."""
    import requests
    # Make an unauthenticated request to see what the server offers
    resp = requests.get(f"{base_url}/", verify=False)
    print(f"Status: {resp.status_code}")

    www_auth = resp.headers.get("WWW-Authenticate", "Not present")
    print(f"WWW-Authenticate: {www_auth}")

    # Parse supported methods
    methods = []
    if "Basic" in www_auth:
        methods.append("Basic")
    if "Negotiate" in www_auth:
        methods.append("Kerberos (Negotiate)")
    if "NTLM" in www_auth:
        methods.append("NTLM")
    if "Bearer" in www_auth:
        methods.append("Bearer / OpenID Connect")

    print(f"Supported auth methods: {', '.join(methods) or 'None detected'}")

    # If Kerberos is available, check the ticket cache
    if "Negotiate" in www_auth:
        import subprocess
        try:
            result = subprocess.run(["klist"], capture_output=True, text=True)
            print(f"\nKerberos ticket cache:\n{result.stdout}")
            if "expired" in result.stdout.lower():
                print("WARNING: Kerberos ticket is expired. Run 'kinit' to refresh.")
        except FileNotFoundError:
            print("\nklist not found. Kerberos tools may not be installed.")

debug_auth("https://your-server/piwebapi")

# Check your current Kerberos tickets
klist

# Get a new ticket (will prompt for password)
kinit username@DOMAIN.COM

# Verify the SPN for PI Web API (run on domain-joined machine)
setspn -L PI-WEB-API-SERVER$

# Check if the correct SPN exists
# Expected: HTTP/pi-web-api-server.domain.com
setspn -Q HTTP/pi-web-api-server.domain.com

# View Kerberos events on Windows (run as admin)
# Event Viewer > Windows Logs > Security
# Filter for Event ID 4768 (TGT request) and 4769 (Service ticket request)

# Enable Kerberos debug logging on Linux
# Add to /etc/krb5.conf under [logging]:
#   default = FILE:/var/log/krb5.log

from requests_ntlm import HttpNtlmAuth

# NTLM requires DOMAIN\username format
session = requests.Session()
session.auth = HttpNtlmAuth("DOMAIN\\username", "password")

# Test the connection
resp = session.get(f"{BASE_URL}/")
print(f"Status: {resp.status_code}")

# Common NTLM issues:
# 1. NTLMv1 disabled by policy: upgrade to NTLMv2 or use Kerberos
# 2. Wrong domain: check with 'whoami /fqdn' on a domain machine
# 3. Account lockout: NTLM retries can lock accounts faster than Kerberos

"""Check which identity PI Web API sees for your connection."""

def check_my_identity(session, base_url):
    """Show the identity PI Web API resolves for your credentials."""
    # The system/userinfo endpoint shows your resolved identity
    resp = session.get(f"{base_url}/system/userinfo")
    if resp.ok:
        info = resp.json()
        print(f"Windows Identity: {info.get('Name', 'Unknown')}")
        print(f"Is Administrator: {info.get('IsAdministrator', False)}")
        print(f"SID: {info.get('SID', 'Unknown')}")
    else:
        print(f"Cannot check identity: {resp.status_code}")

    # Try to list data servers (tests basic PI Data Archive access)
    resp = session.get(f"{base_url}/dataservers")
    if resp.ok:
        servers = resp.json().get("Items", [])
        print(f"\nCan access {len(servers)} data server(s):")
        for s in servers:
            print(f"  - {s['Name']} ({s.get('IsConnected', 'Unknown')})")
    else:
        print(f"\nCannot list data servers: {resp.status_code}")
        if resp.status_code == 403:
            print("  Your PI Identity does not have Read access to any Data Archive.")

check_my_identity(session, BASE_URL)

# Configure separate connect and read timeouts
session.timeout = (10, 120)  # 10s to connect, 120s to read

# Or per-request:
resp = session.get(
    f"{BASE_URL}/streams/{WEB_ID}/recorded",
    params={"startTime": "*-30d", "endTime": "*", "maxCount": 50000},
    timeout=(10, 300),  # 5 minute read timeout for large queries
)

# Use updateOption to control write behavior
response = session.post(
    f"{BASE_URL}/streams/{WEB_ID}/value",
    json={"Value": 42.0, "Timestamp": "2026-03-15T10:00:00Z"},
    params={"updateOption": "Replace"},  # Overwrite existing value
)

# updateOption values:
# "Replace"              - Overwrite existing value at this timestamp
# "Insert"               - Only write if no value exists at this timestamp
# "NoReplace"            - Same as Insert (legacy name)
# "Remove"               - Delete the value at this timestamp
# "InsertNoCompression"  - Insert without applying compression

"""Extract useful details from 500 error responses."""

resp = session.get(f"{BASE_URL}/streams/{WEB_ID}/recorded", params=params)

if resp.status_code == 500:
    try:
        error_body = resp.json()
        # PI Web API usually returns errors in this structure
        errors = error_body.get("Errors", [])
        for err in errors:
            print(f"Server error: {err}")
    except Exception:
        # Sometimes the response is not valid JSON
        print(f"Raw error: {resp.text[:500]}")

"""Step-by-step connection debugging for PI Web API."""
import socket
import requests

def debug_connection(hostname, port=443):
    """Test each layer of connectivity to PI Web API."""

    # Step 1: DNS resolution
    print(f"1. DNS resolution for {hostname}...")
    try:
        ip = socket.gethostbyname(hostname)
        print(f"   Resolved to: {ip}")
    except socket.gaierror as e:
        print(f"   FAILED: {e}")
        print("   Fix: Check hostname spelling. Try IP address directly.")
        return

    # Step 2: TCP connection
    print(f"2. TCP connection to {ip}:{port}...")
    try:
        sock = socket.create_connection((ip, port), timeout=10)
        sock.close()
        print("   Connected successfully")
    except (socket.timeout, ConnectionRefusedError) as e:
        print(f"   FAILED: {e}")
        print("   Fix: Check firewall rules, VPN, or if PI Web API service is running.")
        return

    # Step 3: HTTPS handshake
    print(f"3. HTTPS handshake with {hostname}...")
    try:
        resp = requests.get(
            f"https://{hostname}:{port}/piwebapi/",
            timeout=10,
            verify=False,
            auth=None,  # Skip auth for connectivity test
        )
        print(f"   HTTP {resp.status_code}")
        if resp.status_code == 401:
            print("   Server is reachable and requires authentication (this is good!)")
        elif resp.status_code == 200:
            info = resp.json()
            print(f"   Product: {info.get('ProductTitle', 'Unknown')}")
            print(f"   Version: {info.get('ProductVersion', 'Unknown')}")
    except requests.exceptions.SSLError:
        print("   SSL error (but server is reachable). See SSL section above.")
    except Exception as e:
        print(f"   FAILED: {e}")

debug_connection("your-pi-web-api-server")

"""Comprehensive PI Web API diagnostic script.

Usage:
    python pi_diagnostic.py

Set these environment variables before running:
    PI_WEB_API_URL  = https://your-server/piwebapi
    PI_USERNAME     = DOMAIN\username
    PI_PASSWORD     = your-password
    PI_CA_BUNDLE    = /path/to/cert.pem (optional)
"""
import os
import time
import requests

from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def run_diagnostics():
    base_url = os.environ.get("PI_WEB_API_URL", "https://your-server/piwebapi")
    username = os.environ.get("PI_USERNAME", "")
    password = os.environ.get("PI_PASSWORD", "")
    ca_bundle = os.environ.get("PI_CA_BUNDLE", False)

    session = requests.Session()
    if username:
        session.auth = (username, password)
    session.verify = ca_bundle if ca_bundle else False
    session.headers["X-Requested-With"] = "PiSharp-Diagnostic"

    retry = Retry(total=2, backoff_factor=1, status_forcelist=[502, 503])
    session.mount("https://", HTTPAdapter(max_retries=retry))

    results = {"passed": 0, "failed": 0, "warnings": 0}

    def check(name, func):
        print(f"\n{'='*60}")
        print(f"CHECK: {name}")
        print(f"{'='*60}")
        try:
            func()
            results["passed"] += 1
        except AssertionError as e:
            print(f"  FAILED: {e}")
            results["failed"] += 1
        except Exception as e:
            print(f"  ERROR: {type(e).__name__}: {e}")
            results["failed"] += 1

    def check_connectivity():
        start = time.perf_counter()
        resp = session.get(f"{base_url}/")
        elapsed = time.perf_counter() - start
        assert resp.status_code == 200, f"Expected 200, got {resp.status_code}"
        info = resp.json()
        print(f"  Server: {info.get('ProductTitle', 'Unknown')}")
        print(f"  Version: {info.get('ProductVersion', 'Unknown')}")
        print(f"  Response time: {elapsed:.3f}s")

    def check_data_servers():
        resp = session.get(f"{base_url}/dataservers")
        assert resp.status_code == 200, f"Expected 200, got {resp.status_code}"
        servers = resp.json().get("Items", [])
        assert len(servers) > 0, "No data servers found"
        for s in servers:
            connected = s.get("IsConnected", "Unknown")
            print(f"  {s['Name']}: Connected={connected}")

    def check_af_servers():
        resp = session.get(f"{base_url}/assetservers")
        assert resp.status_code == 200, f"Expected 200, got {resp.status_code}"
        servers = resp.json().get("Items", [])
        print(f"  Found {len(servers)} AF server(s)")
        for s in servers:
            print(f"  {s['Name']}")

    def check_search():
        resp = session.get(
            f"{base_url}/search/query",
            params={"q": "name:sinu*", "count": 5},
        )
        if resp.status_code == 200:
            hits = resp.json().get("Items", [])
            print(f"  Search returned {len(hits)} results")
        elif resp.status_code == 410:
            print("  WARNING: Indexed Search is not enabled (410 Gone)")
            print("  This is OK -- use path-based lookups instead of search.")
            results["warnings"] += 1
        else:
            assert False, f"Search failed with status {resp.status_code}"

    def check_point_read():
        # Try to find and read the sinusoid point
        resp = session.get(f"{base_url}/dataservers")
        if not resp.ok:
            assert False, f"Cannot list data servers: {resp.status_code}"

        servers = resp.json().get("Items", [])
        if not servers:
            print("  SKIP: No data servers available")
            return

        ds_web_id = servers[0]["WebId"]
        resp = session.get(
            f"{base_url}/dataservers/{ds_web_id}/points",
            params={"nameFilter": "sinusoid", "maxCount": 1},
        )
        if resp.ok and resp.json().get("Items"):
            point = resp.json()["Items"][0]
            web_id = point["WebId"]
            print(f"  Found point: {point['Name']}")

            # Read current value
            resp = session.get(
                f"{base_url}/streams/{web_id}/value",
                params={"selectedFields": "Timestamp;Value;Good"},
            )
            assert resp.status_code == 200, f"Read failed: {resp.status_code}"
            val = resp.json()
            print(f"  Current value: {val.get('Value')} at {val.get('Timestamp')}")
            print(f"  Quality good: {val.get('Good', 'Unknown')}")
        else:
            print("  WARNING: sinusoid point not found (may not exist on this server)")
            results["warnings"] += 1

    check("Connectivity & Authentication", check_connectivity)
    check("PI Data Archive Servers", check_data_servers)
    check("PI AF Servers", check_af_servers)
    check("Indexed Search", check_search)
    check("Point Lookup & Data Read", check_point_read)

    print(f"\n{'='*60}")
    print(f"RESULTS: {results['passed']} passed, {results['failed']} failed, {results['warnings']} warnings")
    print(f"{'='*60}")

if __name__ == "__main__":
    run_diagnostics()