> For the complete documentation index, see [llms.txt](https://jacob-taylor.gitbook.io/jacob-taylor-portfolio/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://jacob-taylor.gitbook.io/jacob-taylor-portfolio/detecting-famous-chollima-with-splunk.md).

# Detecting Famous Chollima with Splunk

Author: Jacob Taylor, Security Analyst

<sub>*The demonstrated query relies on Okta logs and the O365 target application. This logic should work as template to create a detection that will meet your own environment's needs.*</sub>

#### Problem

According to [this](https://reports.weforum.org/docs/WEF_Global_Cybersecurity_Outlook_2025.pdf) 2005 Global Cybersecurity Outlook report, 42% of organizations have reported to have fallen victim to a successful social engineering or phishing attack in relation to their remote workers.&#x20;

Commonly, most of these attacks target victim's business email or Single Sign On cloud application access. Gaining access to a business email account can provide a threat actor not only with confidential information but also a way to commit fraud and to perform phishing on other businesses. To accomplish these attacks, the threat actor has to gain unauthorized access. But what happens when a threat actor *is* authorized access?

Threat actor groups, such as North Korea's [Famous Chollima](https://www.huntress.com/threat-library/threat-actors/famous-chollima), use remote work as a way to gain, or be granted, access into businesses. Using stolen identities and deep fake facial filters, the DPRK has perfected a craft to interview for and secure remote IT, Engineering, and Blockchain jobs.&#x20;

After being hired, Famous Chollima doesn't always conduct malicious activity and may "show up" to work, simply collecting a paycheck. But, this group is notoriously known for stealing company data, deploying ransomware, and performing other malicious activity to fund the DPRK governance.

#### Solution

Detecting these fraudulent remote hires can be extremely challenging for organizations, especially for those with minimal security solutions in place. While it should be pushed for your organization to harden the hiring process, it may not always be enough.

Famous Chollima relies on laptop farms being hosted by a US citizen who receives the company computer and allows for the DPRK to connect to it remotely. In order to mask their location, Famous Chollima will often use third party VPN's (Virtual Private Networks) and Proxies when connecting to these computers and accessing the company network.&#x20;

Because of this, security teams can utilize identity log solutions, such as Okta, to track these behaviors. Since Okta is an SSO solution, it will track events such as a user being added to a group. This can be used to your advantage as you determine which, if any, group in your organization is a target for fraudulent, or malicious, remote workers.&#x20;

In this demonstration, the group name **M365\_TECH** is being used and this query is identifying that a new user who was added to this group on `3/03` was observed using a VPN on `3/10` (*presumably the first day the user began working*) 20 times.

<figure><img src="https://3995738258-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2Ft5RYFwPHLHBfpIxm15Gr%2Fuploads%2Fk2e4kfSPtZXFnml3bjrE%2F2.PNG?alt=media&amp;token=433bfb24-9a9c-41cb-86a2-df4c770b5493" alt=""><figcaption></figcaption></figure>

Depending on your environment, this can be considered a high fidelity alert that may reveal the new hire to actually be masquerading their true identity, such as is the case with North Korea's Famous Chollima group infiltrating companies through remote work.

#### Breakdown

**To understand how we get these results, we need analyze what the SPL query is doing**:

1. The query starts in a set search time frame (EX: last 10 minutes) and looks for a user that is operating in or to Okta using a VPN or Proxy. The user name is returned inside quotes along with a result called EMPTY. This empty string is important - if the search does not locate an actual user and does not specify this Empty user, then the next search will run with a wildcard.

{% code overflow="wrap" lineNumbers="true" %}

```spl
index="iam" actor.alternateId IN (*@company.com) debugContext.debugData.tunnels IN (*vpn* *proxy*) client.device=Computer 
| append [stats count | eval actor.alternateId="Empty" | fields actor.alternateId] 
| return 1000 actor.alternateId 
| rex mode=sed field=search "s/(//g" 
| rex mode=sed field=search "s/)//g" 
| rex mode=sed field=search "s/actor.alternateId=//g" 
| rex mode=sed field=search "s/\ OR//g" 
| rex mode=sed field=search "s/""//g"
```

{% endcode %}

<figure><img src="https://3995738258-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2Ft5RYFwPHLHBfpIxm15Gr%2Fuploads%2Fg9dw4UQD8AaqkifMNZZu%2F3.PNG?alt=media&amp;token=8d7e7db4-cc65-4255-aa91-e59f3f552c1e" alt=""><figcaption></figcaption></figure>

2. The previous embedded search returns the user results and applies them to the outer IN ("") section that is looking for target IDs with events related to being added to a specific target app, looking back 45 days. This number could be reduced but it is most likely the case that users will be added to a group prior to their actual start date. 45 days is a generous time frame that allows some wiggle room. If the returned user from the last search is found to have been added to the target app group, then we next learn the date they were added.

{% code overflow="wrap" lineNumbers="true" %}

```spl
index="iam" earliest=-45d displayMessage="Successfully imported new member to an app group" targetAppGroupDisplayName="M365_TECH" target{}.alternateId OR targetUserAlternateId OR targetAppUserAlternateId IN ("Empty" "johndoe@company.com")
| eval time=_time 
| convert ctime(time)
| rex field="time" "(?<TECH_Import_Date>([0-9]{1,2}\/[0-9]{1,2}\/[0-9]{4}))"
| rename targetUserAlternateId as User
| rename targetAppUserAlternateId as User
| rename target{}.alternateId as User
| stats min(TECH_Import_Date) as TECH_Import_Date values(displayMessage) as displayMessage values(targetAppGroupDisplayName) as TargetAppGroup by User
| search User IN (*@company.com)
```

{% endcode %}

<figure><img src="https://3995738258-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2Ft5RYFwPHLHBfpIxm15Gr%2Fuploads%2F679DnmINqImA4HAqKl8m%2F4.PNG?alt=media&amp;token=b7a29594-6b45-4f97-a3fe-ead1a71f7cd1" alt=""><figcaption></figcaption></figure>

3. The next step is to reference the same user(s) from the very first search. This will set the stage to now look for some different criteria without losing our previous search results. To do this, a JOIN function is applied and, once again, we look for a user that has operated in or to Okta using a VPN or Proxy, beginning with a sub-search. The user(s) is returned and utilized in the next outer search. It is important to note that the user from the previous and last search outcome will be joined in the final output.

{% code overflow="wrap" lineNumbers="true" %}

```spl
index="iam" actor.alternateId IN (*@company.com) debugContext.debugData.tunnels IN (*vpn* *proxy*) client.device=Computer
| append [stats count | eval actor.alternateId="Empty" | fields actor.alternateId]
| return 1000 actor.alternateId
```

{% endcode %}

<figure><img src="https://3995738258-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2Ft5RYFwPHLHBfpIxm15Gr%2Fuploads%2Fn7jeGxp04v1W9HA3I8R2%2F5.PNG?alt=media&amp;token=a3c1cd60-3f68-4762-9be7-9c2cf536a455" alt=""><figcaption></figcaption></figure>

4. The user results are now used in a 45 day look back to search for the same criteria of operating in or to Okta using a VPN or Proxy, but collects the Event Count (how many times the activity occurred), acting user agent(s), acting ASN (Autonomous System Number), action(s) or security results such as success & failure, Internet Service provider & IP address, geolocation, VPN or Proxy name, verifying the device type of Computer, first and last time of activity, Okta Apps visited, and first activity time in epoch.

{% code overflow="wrap" lineNumbers="true" %}

```spl
index="iam" earliest=-45d debugContext.debugData.tunnels IN (*vpn* *proxy*) client.device=Computer (actor.alternateId="Empty") OR (actor.alternateId="johndoe@company.com")
| rename "securityContext.asNumber" as ASN, "authenticationContext.externalSessionId" as "Session ID", "actor.alternateId" as User 
| rex field="debugContext.debugData.tunnels" "(?<VPN>(?<=\"\:\").*(?=\"\,\"))"
| eval time=_time 
| convert ctime(time)
| stats dc(time) as EventCount values(user_agent) AS "UserAgent" values(ASN) as ASN values(action) as action values(securityContext.isp) AS ISP values(src_ip) as IP values(request.ipChain{}.geographicalContext.state) as State values(request.ipChain{}.geographicalContext.country) as Country min(time) as first_time max(time) as last_time values(VPN) as VPN values(client.device) as Device values(app) as App min(_time) as first_time_epoch by User
```

{% endcode %}

<figure><img src="https://3995738258-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2Ft5RYFwPHLHBfpIxm15Gr%2Fuploads%2FPLe2qganDNWcqLS76LD6%2F6.PNG?alt=media&amp;token=d03220ea-4e34-49d1-8adf-580f25182af4" alt=""><figcaption></figcaption></figure>

5. This output not only ensures the VPN field is not empty, but is then compared to a time constraint to not re-trigger more than once per user by generating a first and current date while setting a condition for them to be EQUAL, meaning they occurred on the same day. This allows for the triggering condition to be unique and guarantees the behavior is anomalous. Regex is heavily relied upon for this to work.

{% code overflow="wrap" lineNumbers="true" %}

```spl
| eval firsttime=first_time_epoch 
| eval current_time=now()
| convert ctime(current_time) ctime(firsttime)
| rex field="firsttime" "(?<first_date>([0-9]{1,2}\/[0-9]{1,2}\/[0-9]{4}))"
| rex field="current_time" "(?<current_date>([0-9]{1,2}\/[0-9]{1,2}\/[0-9]{4}))"
| where first_date = current_date
| search VPN!=""
```

{% endcode %}

<figure><img src="https://3995738258-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2Ft5RYFwPHLHBfpIxm15Gr%2Fuploads%2FDR2tn3USGQqJkGojzwpu%2F7.PNG?alt=media&amp;token=c711ef37-86f8-4751-8961-624cb756c999" alt=""><figcaption></figcaption></figure>

6. The last step is to table all of the results to provide the alert output. The output reveals that the user was added to the TECH group on 3/3 and that this user was seen using a third party VPN on 3/10 to interact with Okta, O365, Pay & Benefits apps via the company Okta portal.

{% code overflow="wrap" lineNumbers="true" %}

```spl
| table TECH_Import_Date, displayMessage, TargetAppGroup, User, first_time, EventCount, action, App, Device, User, VPN, IP, ISP, State, Country, UserAgent, ASN, last_time
```

{% endcode %}

<figure><img src="https://3995738258-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2Ft5RYFwPHLHBfpIxm15Gr%2Fuploads%2Frjb6wG0bCeXUXKgImN2y%2Fimage.png?alt=media&amp;token=a1e021e3-39a5-4c97-9fde-743029eae25e" alt=""><figcaption></figcaption></figure>

#### Full Query

{% code overflow="wrap" lineNumbers="true" %}

````spl
```Sub-search conducted to look for VPN activity  from a computer device during set search window (recommended to run no less than every 10-15 minutes) from a valid user.``` 
```The user name is returned to the outer search peforming a 45 day lookback to see if the user was recently added to TECH group. If so, the search will locate the import date.```
index="iam" earliest=-45d displayMessage="Successfully imported new member to an app group" targetAppGroupDisplayName="M365_TECH" target{}.alternateId OR targetUserAlternateId OR targetAppUserAlternateId IN ([search index="iam" actor.alternateId IN (*@company.com) debugContext.debugData.tunnels IN (*vpn* *proxy*) client.device=Computer 
```The Appended stats are used to prevent the search from returning an empty value, throwing off the main search results by searching with a wildcard due to the empty result.```
| append [stats count | eval actor.alternateId="Empty" | fields actor.alternateId]
| return 1000 actor.alternateId
| rex mode=sed field=search "s/\(//g" 
| rex mode=sed field=search "s/\)//g" 
| rex mode=sed field=search "s/actor.alternateId\=//g" 
| rex mode=sed field=search "s/\\ OR//g"
| rex mode=sed field=search "s/\"\"//g"])
| eval time=_time 
| convert ctime(time)
| rex field="time" "(?<TECH_Import_Date>([0-9]{1,2}\/[0-9]{1,2}\/[0-9]{4}))"
| rename targetUserAlternateId as User
| rename targetAppUserAlternateId as User
| rename target{}.alternateId as User
| stats min(TECH_Import_Date) as TECH_Import_Date values(displayMessage) as displayMessage values(targetAppGroupDisplayName) as TargetAppGroup by User
```Ensure search is continuing on a legitimate user.```
| search User IN (*@company.com)
```Join used to add previous group import information to next search by joining the User field in both searches.```
| join [search index="iam" earliest=-45d debugContext.debugData.tunnels IN (*vpn* *proxy*) client.device=Computer ([search index="iam" actor.alternateId IN (*@company.com) debugContext.debugData.tunnels IN (*vpn* *proxy*) client.device=Computer
```The Appended stats are used to prevent the search from returning an empty value, throwing off the main search results by searching with a wildcard due to the empty result.```
| append [stats count | eval actor.alternateId="Empty" | fields actor.alternateId]
| return 1000 actor.alternateId])
| rename "securityContext.asNumber" as ASN, "authenticationContext.externalSessionId" as "Session ID", "actor.alternateId" as User 
```Regex is used to extract the VPN name from the raw data and to create a field called VPN```
| rex field="debugContext.debugData.tunnels" "(?<VPN>(?<=\"\:\").*(?=\"\,\"))"
| eval time=_time 
| convert ctime(time)
| stats dc(time) as EventCount values(user_agent) AS "UserAgent" values(ASN) as ASN values(action) as action values(securityContext.isp) AS ISP values(src_ip) as IP values(request.ipChain{}.geographicalContext.state) as State values(request.ipChain{}.geographicalContext.country) as Country min(time) as first_time max(time) as last_time values(VPN) as VPN values(client.device) as Device values(app) as App min(_time) as first_time_epoch by User
```Time constraint set to not re-trigger more than time per user in the event legitimate activity is continuing.```
| eval firsttime=first_time_epoch 
| eval current_time=now()
| convert ctime(current_time) ctime(firsttime)
| rex field="firsttime" "(?<first_date>([0-9]{1,2}\/[0-9]{1,2}\/[0-9]{4}))"
| rex field="current_time" "(?<current_date>([0-9]{1,2}\/[0-9]{1,2}\/[0-9]{4}))"
| where first_date = current_date
```Ensure search is continuing only on VPN activity by user.```
| search VPN!=""]
```Output results from both searches if User is the same.```
| table TECH_Import_Date, displayMessage, TargetAppGroup, User, first_time, EventCount, action, App, Device, User, VPN, IP, ISP, State, Country, UserAgent, ASN, last_time
````

{% endcode %}