Help with EQL Rule to Detect Unauthorized State Transitions for Traffic Lights

Elastic Security
1

Hi Elastic Community,

I’m currently working on a project where I need to detect unauthorized state transitions for traffic lights in a real-time event stream using Elastic EQL. I’ve already set up the state machine and wrote the EQL query to detect the bad transitions, but I’m encountering some difficulties, especially in fine-tuning the rule and detecting transitions that should not occur.

Here is the state machine :

image
image233×417 16.1 KB

Here is the state table :

image
image622×207 12.6 KB

Here is my definition of the rule in EQL:

sequence
  [ any where "event.light state" == "dark" or "event.light state" == "stop-Then-Proceed" or "event.light state" == "permissive-clearance" ]
  [ any where not "event.light state" == "stop-And-Remain" ]
  
  [ any where "event.light state" == "stop-And-Remain" ]
  [ any where not "event.light state" == "pre-Movement"]
  
  [ any where "event.light state" == "pre-movement" ]
  [ any where not "event.light state" == "permissive-Movement-Allowed" or "event.light state" == "caution-Conflicting-Traffic"]
 
  
  [ any where   "event.light state" == "permissive-Movement-Allowed" or "event.light state" == "caution-Conflicting-Traffic"]
  [ any where not "event.light state" == "permissive-clearance" ]

Thanks for you help !

1 Like
2

Hi @Kuly2Fraise ! This is an interesting example, thanks for the helpful table and diagram!

Reading through your EQL query, each pair of [allowed_previous_state, NOT current_state] sequences makes sense: encountering those two events in sequence would define an unauthorized state transition. However, I see a few issues with the current query:

  1. Your query lists 8 events that must occur, in sequence, for a "hit" to be found. I suspect what you want instead is four individual queries, each looking for a single bad transition.

  2. Taking this piece as an example:

    [ any where "event.light state" == "stop-And-Remain" ]
[ any where not "event.light state" == "pre-Movement"]

    Your query is looking for e.g. any stop-and-remain event, followed by any NOT pre-movement event. But what if your data involves multiple traffic lights (as I imagine it will)? If a light at one end of town is stop-and-remain, and then a light on the other side of town enters stop-and-remain, that would be caught by this sequence. The by keyword should allow you to apply your search to only a single traffic camera at a time.

  3. Using the same example as above: what happens if you (intentionally or accidentally) search non-light-state data? In that situation, NOT pre-movement would match nearly everything (since the absence of light state is also NOT pre-movement), and cause a lot of false positives. To address this you'll need to filter your data somehow, either with additional where conditions or (ideally) by using an appropriate event category field. I can't make a specific recommendation without seeing your data, though.

I hope these tips get you on the right track. If they don't, or you have additional issues, please share more specifics about where you're at:

  1. Examples of data being queried on
  2. The query being used
  3. The actual response
  4. The expected response

P.S. One more quick suggestion: check out the ECS Guidelines & Best Practices to get your data in an optimal shape for elasticsearch. We already touched on the use of event.category to replace the anys you have now, but e.g. the recommendation of "combine words using underscore" would prevent you from having to quote light state all over the place :wink: .

2 Likes
3

Hello @RylandHerrick ,

Thank you for you answer.

I've tried to modify my EQL query according to your recommendation, but I get an ‘Unsupported join key’ error on the ‘by’ with the event.wlan-src field which exists in the packages. For your information, the unique field that identifies the traffic light is ‘event.wlan-src’, which is why I wanted to do the join on that field. Am I doing something wrong?

sequence by event.wlan-src
  [ any where event.light_state in ("dark", "stop-Then-Proceed", "permissive-clearance") ]
  [ any where event.light_state != "stop-And-Remain" ]

Here is my json document coming from my traffic light:

{
  "_index": ".ds-filebeat-9.0.0-2024.12.06-000002",
  "_id": "0nuFzpMBHKffx07uVSvf",
  "_version": 1,
  "_score": 0,
  "_source": {
    "@timestamp": "2024-12-16T08:11:03.380Z",
    "log": {
      "offset": 4626092,
      "file": {
        "path": "/home/user/workspace/traffic_light_state.json"
      },
      "original": "Traffic light state",
      "level": "info",
      "logger": "traffic_light_state",
      "origin": {
        "function": "<module>",
        "file": {
          "name": "print-traffic-light-state.py",
          "line": 52
        }
      }
    },
    "process": {
      "thread": {
        "name": "MainThread",
        "id": 1996342032
      },
      "name": "MainProcess",
      "pid": 23495
    },
    "ecs": {
      "version": "1.6.0"
    },
    "message": "Traffic light state",
    "event": {
      "wlan-src": "00:0d:41:12:19:78",
      "MID": "00:0d:41:12:19:78",
      "light_state": "permissive-Movement-Allowed",
      "timestamp": "2024-12-16T08:11:03.375923"
    },
    "input": {
      "type": "log"
    },
    "agent": {
      "type": "filebeat",
      "version": "9.0.0",
      "ephemeral_id": "ed18c6ea-1919-4590-a503-c852ca6f3e94",
      "id": "59d71abc-5da2-4ba8-a21b-a779a9d8e026",
      "name": "MK5-RSU"
    },
    "host": {
      "name": "mk5-rsu",
      "os": {
        "kernel": "4.14.98-00009-g815aa81f1",
        "codename": "focal",
        "type": "linux",
        "platform": "ubuntu",
        "version": "20.04.5 LTS (Focal Fossa)",
        "family": "debian",
        "name": "Ubuntu"
      },
      "id": "4db9660272c5a45f2a85d922631a7952",
      "containerized": false,
      "ip": [
        "160.98.26.181",
        "fe80::6e5:48ff:fe10:c924",
        "10.1.1.3",
        "fe80::b0b0:9dff:fed6:299e"
      ],
      "mac": [
        "00-44-4F-54-33-00",
        "00-44-4F-54-34-00",
        "02-24-31-2D-A2-39",
        "04-E5-48-10-C9-24",
        "04-E5-48-10-C9-25",
        "0A-3F-71-A3-3F-E3",
        "12-BA-D7-BC-A7-C6",
        "2E-FB-69-24-41-BE",
        "32-24-5A-72-23-AA",
        "36-85-E7-26-0C-CD",
        "4A-3A-1B-E6-61-B3",
        "4A-5A-71-2B-EE-34",
        "4E-8C-5A-A6-D2-90",
        "62-06-40-44-30-A7",
        "6E-DB-5C-1E-72-EA",
        "8A-44-CF-75-9D-33",
        "8A-CA-87-0A-CB-C9",
        "A6-F7-72-FC-58-9C",
        "B2-B0-9D-D6-29-9E",
        "BA-32-AF-D2-D4-72",
        "BA-CA-9B-D6-49-B3",
        "CE-42-F6-F9-9C-35",
        "DE-ED-D7-61-2B-F1",
        "E2-27-6A-43-AF-2D",
        "E6-17-C9-45-E3-1E",
        "EA-59-31-68-D0-D9"
      ],
      "hostname": "MK5-RSU",
      "architecture": "armv7l"
    }
  },
  "fields": {
    "process.name.text": [
      "MainProcess"
    ],
    "host.os.name.text": [
      "Ubuntu"
    ],
    "host.hostname": [
      "MK5-RSU"
    ],
    "process.pid": [
      23495
    ],
    "host.mac": [
      "00-44-4F-54-33-00",
      "00-44-4F-54-34-00",
      "02-24-31-2D-A2-39",
      "04-E5-48-10-C9-24",
      "04-E5-48-10-C9-25",
      "0A-3F-71-A3-3F-E3",
      "12-BA-D7-BC-A7-C6",
      "2E-FB-69-24-41-BE",
      "32-24-5A-72-23-AA",
      "36-85-E7-26-0C-CD",
      "4A-3A-1B-E6-61-B3",
      "4A-5A-71-2B-EE-34",
      "4E-8C-5A-A6-D2-90",
      "62-06-40-44-30-A7",
      "6E-DB-5C-1E-72-EA",
      "8A-44-CF-75-9D-33",
      "8A-CA-87-0A-CB-C9",
      "A6-F7-72-FC-58-9C",
      "B2-B0-9D-D6-29-9E",
      "BA-32-AF-D2-D4-72",
      "BA-CA-9B-D6-49-B3",
      "CE-42-F6-F9-9C-35",
      "DE-ED-D7-61-2B-F1",
      "E2-27-6A-43-AF-2D",
      "E6-17-C9-45-E3-1E",
      "EA-59-31-68-D0-D9"
    ],
    "log.logger": [
      "traffic_light_state"
    ],
    "host.ip": [
      "160.98.26.181",
      "fe80::6e5:48ff:fe10:c924",
      "10.1.1.3",
      "fe80::b0b0:9dff:fed6:299e"
    ],
    "agent.type": [
      "filebeat"
    ],
    "event.light_state": [
      "permissive-Movement-Allowed"
    ],
    "host.os.version": [
      "20.04.5 LTS (Focal Fossa)"
    ],
    "host.os.kernel": [
      "4.14.98-00009-g815aa81f1"
    ],
    "host.os.name": [
      "Ubuntu"
    ],
    "event.MID": [
      "00:0d:41:12:19:78"
    ],
    "log.level": [
      "info"
    ],
    "agent.name": [
      "MK5-RSU"
    ],
    "host.name": [
      "mk5-rsu"
    ],
    "host.id": [
      "4db9660272c5a45f2a85d922631a7952"
    ],
    "log.original": [
      "Traffic light state"
    ],
    "process.thread.name": [
      "MainThread"
    ],
    "log.origin.file.line": [
      52
    ],
    "host.os.type": [
      "linux"
    ],
    "event.wlan-src": [
      "00:0d:41:12:19:78"
    ],
    "host.os.codename": [
      "focal"
    ],
    "input.type": [
      "log"
    ],
    "log.offset": [
      4626092
    ],
    "agent.hostname": [
      "MK5-RSU"
    ],
    "event.timestamp": [
      "2024-12-16T08:11:03.375923"
    ],
    "message": [
      "Traffic light state"
    ],
    "host.architecture": [
      "armv7l"
    ],
    "process.name": [
      "MainProcess"
    ],
    "@timestamp": [
      "2024-12-16T08:11:03.380Z"
    ],
    "log.origin.file.name": [
      "print-traffic-light-state.py"
    ],
    "log.origin.function": [
      "<module>"
    ],
    "agent.id": [
      "59d71abc-5da2-4ba8-a21b-a779a9d8e026"
    ],
    "ecs.version": [
      "1.6.0"
    ],
    "host.containerized": [
      false
    ],
    "host.os.platform": [
      "ubuntu"
    ],
    "log.file.path": [
      "/home/user/workspace/traffic_light_state.json"
    ],
    "agent.ephemeral_id": [
      "ed18c6ea-1919-4590-a503-c852ca6f3e94"
    ],
    "agent.version": [
      "9.0.0"
    ],
    "host.os.family": [
      "debian"
    ],
    "process.thread.id": [
      1996342032
    ]
  }
}

The current response is empty. :frowning:

The expected behavior is to detect incorrect transitions from the previous state. If I am currently in the "Stop-And-Remain" state, I need to check that the previous state was either "dark," "permissive-clearance," or "Stop-Then-Proceed."

Thanks :slight_smile:

4

@Kuly2Fraise this may be due to the hyphen in the field name (best practices:wink: ); try escaping the fieldname with backticks.

Regarding the "empty response:"

  1. Can you please share the full query that you're using?
  2. Have you eliminated any broader issues like missing/conflicting mappings? Are you able to e.g. retrieve results with sequence [any where event.light_state != null] [any where event.light_state != null]
  3. Can you verify that the data contains the sequence that you're looking for? Remember that they have to be sequential by @timestamp, and (now) contain the same event.wlan-src key.
1 Like
5

@RylandHerrick

I have good news! My rule is working and successfully generates alerts when a sequence doesn't meet the criteria, as shown in the following image:

image
image1345×432 76.3 KB

However, I've noticed a significant number of false positives, likely because I haven't accounted for all the red light transition states. Are we certain that Kibana always selects the packet closest to the first one in the sequence? I suspect it might be considering packets further away, which could be causing these false positives. Here's an example to explain my point:

  • P1: stop-And-Remain
  • P2: dark
  • P3: other_state

In this case, Kibana may be checking the sequence between packets 1 and 3, instead of packets 1 and 2, which is likely causing the issue.

Could you assist me in correcting this?

sequence by `event.wlan-src`
  [any where event.light_state not in ("dark", "stop-Then-Proceed", "permissive-clearance", "stop-And-Remain", "protected-clearance")]
  [any where event.light_state == "stop-And-Remain"]
6

Excellent! If it might help others coming to this post, please share what steps got you there!

Definitely not; any and all N events in your search corpus matching the specified sequence will be considered "hits" by EQL. So in your example it would definitely find both [P1, P2] AND [P1, P3].

I suspect you'll want to use with maxspan to constrain your sequences to to a certain timeframe. However, with maxspan represents "only this close," which isn't exactly the same as "closest," so there's probably still a possibility of false positives there. Depending on the shape of your data, you may be able to minimize this with additional filters.

Please share any other findings you have, and remember to mark this as "solved" once you're satisfied. Thanks!

7

@RylandHerrick,

Thanks for your help! To achieve my goal, I simply added "with maxspan=2s" to prevent the rule from looking further than necessary, which helps avoid inconsistent results. Here’s the final EQL code for the red light state:

sequence by `event.wlan-src` with maxspan=2s
  [any where event.light_state not in ("dark", "stop-Then-Proceed", "permissive-clearance", "stop-And-Remain", "protected-clearance")]
  [any where event.light_state == "stop-And-Remain"]

Now, the rule works as expected when I send incorrect packets -> Alert generated. And under normal conditions (no error), no alerts are triggered.

Thanks again for your help, I’ll close this thread.

Happy holidays and best wishes for the new year! :slight_smile:

1 Like