fix: correct CWMP session management side effects
- state_machine_manager: add deviceToRemotes reverse index (Map<deviceKeyJson, Set<remoteKey>>) so completeMachine/failMachine clean up remoteToDevice in O(1) — fixes unbounded memory leak - state_machine_manager: getHashIdForRemoteIp() now only matches RUNNING sessions and returns undefined when multiple candidates exist (NAT/CGNAT safety) - state_machine_manager: IPv6 bracket normalization in IP comparison - cwmp_controller: apply IP fallback to TransferComplete and generic response handler (same new-TCP-port scenario as empty POST after firmware reboot) - cwmp_logger_middleware: reuse request.cwmp.sessionId instead of recomputing from socket to avoid formula drift
This commit is contained in:
1 parent
d44d896556
commit
7835f99657
3 files changed
+78
-23
No files matched your search
@@ -81,6 +81,7 @@ export default class CwmpController {
|
||||
|
||||
if (message?.type === 'TransferComplete') {
|
||||
const hashId = manager.getHashIdForRemote(remoteKey)
|
||||
?? manager.getHashIdForRemoteIp(request.ip())
|
||||
if (hashId) ctx.cwmpDeviceId = hashId
|
||||
logger.debug({ sessionId, hashId }, 'CWMP sending TransferCompleteResponse')
|
||||
const resp = buildSoapEnvelope(
|
||||
@@ -92,6 +93,7 @@ export default class CwmpController {
|
||||
|
||||
manager.ensureSessionForRemote(remoteKey, sessionId)
|
||||
const hashId = manager.getHashIdForRemote(remoteKey)
|
||||
?? manager.getHashIdForRemoteIp(request.ip())
|
||||
if (hashId) ctx.cwmpDeviceId = hashId
|
||||
|
||||
if (message && manager.hasSession(sessionId)) {
|
||||
|
||||
@@ -34,12 +34,15 @@ export default class CwmpLoggerMiddleware {
|
||||
const { request, response } = ctx
|
||||
const rawBody = request.raw() || ''
|
||||
|
||||
const socket = request.request?.socket
|
||||
const remoteIp = request.ip()
|
||||
const remotePort = socket?.remotePort || 0
|
||||
const localAddress = socket?.localAddress || 'unknown'
|
||||
const localPort = socket?.localPort || 0
|
||||
const sessionId = `${remoteIp}:${remotePort}->${localAddress}:${localPort}`
|
||||
// Reuse sessionId already computed by CwmpMiddleware to avoid drift
|
||||
const sessionId = request.cwmp?.sessionId ?? (() => {
|
||||
const socket = request.request?.socket
|
||||
const remoteIp = request.ip()
|
||||
const remotePort = socket?.remotePort || 0
|
||||
const localAddress = socket?.localAddress || 'unknown'
|
||||
const localPort = socket?.localPort || 0
|
||||
return `${remoteIp}:${remotePort}->${localAddress}:${localPort}`
|
||||
})()
|
||||
|
||||
let capturedResponseXml: string | null = null
|
||||
let isEmptyResponse = false
|
||||
|
||||
@@ -40,7 +40,8 @@ export class StateMachineManager extends EventEmitter {
|
||||
private machines: Map<string, StateMachineInstance> = new Map()
|
||||
private deviceSessions: Map<string, string> = new Map() // deviceKey -> sessionId
|
||||
private sessionMachines: Map<string, string> = new Map() // sessionId -> machineId
|
||||
private remoteToDevice: Map<string, string> = new Map() // remoteKey -> deviceKey
|
||||
private remoteToDevice: Map<string, string> = new Map() // remoteKey -> deviceKeyJson
|
||||
private deviceToRemotes: Map<string, Set<string>> = new Map() // deviceKeyJson -> Set<remoteKey>
|
||||
private cleanupInterval: ReturnType<typeof setInterval> | null = null
|
||||
|
||||
private static instance: StateMachineManager | null = null
|
||||
@@ -221,10 +222,31 @@ export class StateMachineManager extends EventEmitter {
|
||||
}
|
||||
|
||||
/**
|
||||
* Bind a remote endpoint to a device for session rebinding
|
||||
* Bind a remote endpoint to a device for session rebinding.
|
||||
* Maintains the reverse index deviceToRemotes for O(1) cleanup.
|
||||
*/
|
||||
bindRemoteToDevice(remoteKey: string, deviceId: DeviceId): void {
|
||||
this.remoteToDevice.set(remoteKey, JSON.stringify(deviceId))
|
||||
const deviceKeyJson = JSON.stringify(deviceId)
|
||||
this.remoteToDevice.set(remoteKey, deviceKeyJson)
|
||||
let remotes = this.deviceToRemotes.get(deviceKeyJson)
|
||||
if (!remotes) {
|
||||
remotes = new Set()
|
||||
this.deviceToRemotes.set(deviceKeyJson, remotes)
|
||||
}
|
||||
remotes.add(remoteKey)
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove all remoteToDevice entries for a device. Called on session completion/failure.
|
||||
*/
|
||||
private unbindRemotesForDevice(deviceKeyJson: string): void {
|
||||
const remotes = this.deviceToRemotes.get(deviceKeyJson)
|
||||
if (remotes) {
|
||||
for (const rk of remotes) {
|
||||
this.remoteToDevice.delete(rk)
|
||||
}
|
||||
this.deviceToRemotes.delete(deviceKeyJson)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -249,28 +271,54 @@ export class StateMachineManager extends EventEmitter {
|
||||
}
|
||||
|
||||
/**
|
||||
* Fallback lookup: returns the hash identifier for any active device whose
|
||||
* Fallback lookup: returns the hash identifier for a RUNNING device whose
|
||||
* last known remote IP matches the given IP (ignoring port).
|
||||
* Used when a CPE opens a new TCP connection for an empty POST and the
|
||||
* exact remoteKey (IP:port) is not in the map.
|
||||
*
|
||||
* Only considers devices with an active (RUNNING) state machine to avoid
|
||||
* returning a stale or wrong device under NAT/CGNAT scenarios.
|
||||
* Returns undefined if zero or multiple RUNNING devices match the IP.
|
||||
*/
|
||||
getHashIdForRemoteIp(remoteIp: string): string | undefined {
|
||||
const candidates: string[] = []
|
||||
|
||||
for (const [key, deviceKeyJson] of this.remoteToDevice.entries()) {
|
||||
const keyIp = key.split(':').slice(0, -1).join(':') // strip port (handles IPv6 too)
|
||||
if (keyIp !== remoteIp) continue
|
||||
try {
|
||||
const deviceId = JSON.parse(deviceKeyJson)
|
||||
return getDeviceHashIdentifier({
|
||||
manufacturer: deviceId.manufacturer ?? '',
|
||||
oui: deviceId.oui,
|
||||
productClass: deviceId.productClass,
|
||||
serialNumber: deviceId.serialNumber,
|
||||
})
|
||||
} catch {
|
||||
return undefined
|
||||
// Strip port: for IPv4 "1.2.3.4:port" → "1.2.3.4"
|
||||
// for IPv6 "[::1]:port" → "[::1]" (request.ip() returns without brackets)
|
||||
const lastColon = key.lastIndexOf(':')
|
||||
const keyIp = lastColon > 0 ? key.slice(0, lastColon) : key
|
||||
// Normalize: strip brackets if present so "[::1]" matches "::1"
|
||||
const normalizedKeyIp = keyIp.replace(/^\[|\]$/g, '')
|
||||
if (normalizedKeyIp !== remoteIp) continue
|
||||
|
||||
// Only consider devices with a currently RUNNING session
|
||||
const existingSession = this.deviceSessions.get(deviceKeyJson)
|
||||
if (!existingSession) continue
|
||||
const machineId = this.sessionMachines.get(existingSession)
|
||||
if (!machineId) continue
|
||||
const machine = this.machines.get(machineId)
|
||||
if (!machine || machine.status !== MachineStatus.RUNNING) continue
|
||||
|
||||
if (!candidates.includes(deviceKeyJson)) {
|
||||
candidates.push(deviceKeyJson)
|
||||
}
|
||||
}
|
||||
return undefined
|
||||
|
||||
// Ambiguous under NAT/CGNAT: refuse to guess
|
||||
if (candidates.length !== 1) return undefined
|
||||
|
||||
try {
|
||||
const deviceId = JSON.parse(candidates[0])
|
||||
return getDeviceHashIdentifier({
|
||||
manufacturer: deviceId.manufacturer ?? '',
|
||||
oui: deviceId.oui,
|
||||
productClass: deviceId.productClass,
|
||||
serialNumber: deviceId.serialNumber,
|
||||
})
|
||||
} catch {
|
||||
return undefined
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -470,6 +518,7 @@ export class StateMachineManager extends EventEmitter {
|
||||
const deviceKey = JSON.stringify(machine.deviceId)
|
||||
this.deviceSessions.delete(deviceKey)
|
||||
this.sessionMachines.delete(machine.session)
|
||||
this.unbindRemotesForDevice(deviceKey)
|
||||
// Drain any leftover tasks so they don't bleed into the next session
|
||||
this.taskQueue.clearPendingTasksForDevice(machine.deviceId)
|
||||
}
|
||||
@@ -480,6 +529,7 @@ export class StateMachineManager extends EventEmitter {
|
||||
const deviceKey = JSON.stringify(machine.deviceId)
|
||||
this.deviceSessions.delete(deviceKey)
|
||||
this.sessionMachines.delete(machine.session)
|
||||
this.unbindRemotesForDevice(deviceKey)
|
||||
logger.error(`State machine failed: ${error}`)
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user