1
00:00:02,180 --> 00:00:03,800
So, hello everyone.

2
00:00:03,800 --> 00:00:05,380
Thank you for tuning in.

3
00:00:05,460 --> 00:00:07,680
Today, we're going to talk about Apache Spark.

4
00:00:07,840 --> 00:00:14,820
Now, I will just assume that while some of you may be familiar with the technology, a lot of you did not have actual chance to play with the cluster.

5
00:00:14,820 --> 00:00:17,040
So, we'll take it right from the beginning.

6
00:00:17,840 --> 00:00:29,220
So yeah, we'll see how Spark works, what makes it so popular, how to view some common and default settings, and obviously, some bones along the way to bypass authentication and execute code, right?

7
00:00:29,360 --> 00:00:30,680
So, let's get going, right?

8
00:00:30,680 --> 00:00:44,480
So, our story takes us back to 2008, when the economy was not much better than today's, but Internet Explorer was still a thing, and the big tech companies were doing what they did best, quietly collecting the world's data.

9
00:00:44,560 --> 00:00:52,860
When you gather that much data, and I'm talking about terabytes and terabytes of data, performing even the simplest calculations becomes quite challenging.

10
00:00:53,020 --> 00:01:03,300
Now, you can either get a mainframe for half a million dollars, plus whatever it takes to get an intern to work on it, or you can actually distribute your files over thousands and thousands of servers.

11
00:01:03,520 --> 00:01:10,360
That way, your computations are small, though multiplied, and that's one way to handle the load.

12
00:01:10,360 --> 00:01:15,980
And the standard way of doing so, I would say, is by using the Apache Hadoop framework.

13
00:01:16,420 --> 00:01:22,020
Now, if you've ever worked with Hadoop, you must know that it's actually quite a hostile environment.

14
00:01:22,140 --> 00:01:24,200
It's a very complex environment.

15
00:01:24,300 --> 00:01:38,540
So, first of all, just to give you a quick example, you need to fragment your files using the HDFS file system, so you end up with multiple fragments on multiple servers, these files, and then you need to do coherent computations on these small fragments,

16
00:01:38,540 --> 00:01:40,280
so you need the MapReduce framework.

17
00:01:40,340 --> 00:01:50,640
But then you end up with thousands of processes running on multiple machines, so you need a way to schedule them, enter YARN, but then you want to do some SQL querying, so you need Apache Drill, et cetera, et cetera.

18
00:01:50,640 --> 00:02:00,560
So, you end up with this plethora of tools, each with their own latencies, with their own learning curve, et cetera, just to cover all your needs in data analytics.

19
00:02:00,560 --> 00:02:06,020
So , just to give you an example, this is what it takes to do a simple word count on MapReduce.

20
00:02:06,020 --> 00:02:07,420
So, this is the first page of Java.

21
00:02:07,420 --> 00:02:12,520
This is the second page, which is downright outrageous for a data processing solution, right?

22
00:02:13,360 --> 00:02:25,880
So, in our story, we're around 2009, and there's this guy named Matei Azaria, who was at Berkeley, I believe, at the time, who I imagine looked at this code and went, are you fucking crazy?

23
00:02:25,880 --> 00:02:28,340
That's not going to work.

24
00:02:28,560 --> 00:02:30,600
A word count is a simple operation.

25
00:02:30,600 --> 00:02:34,220
A word count should be done in five lines of code.

26
00:02:34,500 --> 00:02:35,960
Apache Spark was born.

27
00:02:36,360 --> 00:02:37,620
Now, this is not the official story.

28
00:02:37,620 --> 00:02:45,160
The official story is that he was working on something called Mesos, which is a resource manager, and he invented Spark as a proof of concept, but I like my version better.

29
00:02:45,340 --> 00:02:57,320
So, anyway, needless to say, we'll go through the code later on, but right off the bat, you can see that you can do pretty much the same processing that you would do in Apache Hadoop, except that you can do it at 10 times less code than Spark.

30
00:02:57,560 --> 00:02:59,820
But it's also much faster.

31
00:03:00,280 --> 00:03:04,340
So, it's three times faster on 10 times less nodes, so it's 30 times faster.

32
00:03:04,340 --> 00:03:10,300
You see, Hadoop will flush everything to disk whenever it gets a chance, whereas Spark will try to keep everything in memory.

33
00:03:10,300 --> 00:03:14,560
It was developed in an era where memory was getting cheaper and cheaper, so it kind of makes sense.

34
00:03:14,560 --> 00:03:24,540
But if you're running in a cloud environment where you pay for the usage, having something that runs 30 times faster means that it's 30 times cheaper, actually.

35
00:03:24,900 --> 00:03:33,320
And that partly explains the gigantic boom and popularity of Spark amongst the big data companies.

36
00:03:33,320 --> 00:03:41,180
If you're serious about your big data processing and cluster front, you will be using Spark for at least some part of it.

37
00:03:41,500 --> 00:03:49,500
And so, yeah, Spark boomed, was adopted by the Apache Foundation, and it grew to encompass many aspects of data analytics.

38
00:03:49,500 --> 00:03:53,860
If you're gonna do SQL, streaming, machine learning, graph processing, you know, it got you covered.

39
00:03:54,180 --> 00:03:59,320
So it's developed half in Java, half in Scala, but it also has some connectors in Python and R.

40
00:03:59,320 --> 00:04:01,580
So , yeah, awesome.

41
00:04:01,640 --> 00:04:08,360
So basically, when you have this powerful framework, suddenly, you know, everything looks like a nail.

42
00:04:08,360 --> 00:04:16,280
So if you're gonna do revenue prediction, you just hook Spark to your, I don't know, Redshift cluster, download those financial statements, and then, you know, you can do yourself.

43
00:04:16,280 --> 00:04:26,560
If you're gonna do fraud analysis, user targeting, same, you hook it up to your Cassandra database, your Astra buckets, you, you know, upload or download those log files, and then you do your computation.

44
00:04:26,560 --> 00:04:35,460
So basically, what I'm saying is that Apache Spark is often in the, at the junction of almost every important data store inside the company.

45
00:04:35,700 --> 00:04:46,480
And that's what made it interesting, at least for me, because, well, if you can access this tool, you have pretty much, you're pretty much open to every database that's out there.

46
00:04:46,480 --> 00:04:48,380
So, you know, how is it protected?

47
00:04:48,380 --> 00:04:49,240
How does it work?

48
00:04:49,240 --> 00:04:50,920
What is its security, et cetera?

49
00:04:51,020 --> 00:05:05,120
And so what I did, I did what I, any sensible person would do, I went to the documentation, to their website, and I went to the security page, and lo and behold, from this magnificent sentence that says security in Spark is off by default, well, fuck.

50
00:05:06,580 --> 00:05:07,800
And what does it do again?

51
00:05:07,800 --> 00:05:16,420
Oh yeah, it's this gigantic framework cluster of hundreds and hundreds of machines that have access to every database inside the company.

52
00:05:16,420 --> 00:05:17,300
And what does it do?

53
00:05:17,300 --> 00:05:24,120
Oh yeah, it does distributed data processing, or as I'd like to call it, it's really distributed remote code execution.

54
00:05:24,120 --> 00:05:25,660
So I thought this would be fun.

55
00:05:26,120 --> 00:05:32,300
And I hope I piqued your interest because really that's what sparked my, the whole endeavor, no pun intended.

56
00:05:32,620 --> 00:05:43,220
And I hope I got you interested enough to actually continue along this presentation so we can explore how it works and some bones that I found, right?

57
00:05:43,220 --> 00:05:49,380
So, oh yeah, in keeping with DEF CON's tradition, I think it's time for drinking.

58
00:05:50,500 --> 00:05:54,240
I should not do this online, but oh, what the heck?

59
00:05:55,260 --> 00:05:57,760
So to DEF CON safe mode, cheers.

60
00:06:01,360 --> 00:06:02,920
Yeah, that will kick in later.

61
00:06:02,920 --> 00:06:06,880
Anyway, so what is Spark and how does it work?

62
00:06:07,320 --> 00:06:10,380
I hope it's going to be one take because I cannot do that five times.

63
00:06:10,460 --> 00:06:12,000
So how does it work?

64
00:06:12,000 --> 00:06:18,220
So as in everything in distributed processing, basically we start with a bunch of machines that we're going to call workers.

65
00:06:18,860 --> 00:06:24,880
Now these workers are just basically brainless machines that will execute whatever is sent their way.

66
00:06:24,880 --> 00:06:33,940
Now on each worker, you find a JVM process, so a Spark process that's called an executor and that's going to do the actual execution.

67
00:06:34,440 --> 00:06:42,040
Now you can have many executors on each worker and it's the basically executor that defines your power in terms of parallelism.

68
00:06:42,040 --> 00:06:49,740
So if you have three, you have 10 workers on each worker or each machine, basically you have three executors so you can run 30 tasks in parallel.

69
00:06:51,080 --> 00:07:01,420
So yeah, you have like some default HTTP port that gives a status of what's going on on the machine you're not interested in, but you have the random RPC port that is set for each executor.

70
00:07:01,420 --> 00:07:09,080
And it's this one that the rest of the cluster contacts in order to send and receive a status from the worker.

71
00:07:10,020 --> 00:07:15,920
Now the worker is fine, but the most important piece or one of the most important pieces, the cluster manager, and that's our second piece.

72
00:07:16,500 --> 00:07:20,660
And the cluster manager's sole job is to actually schedule the application.

73
00:07:20,660 --> 00:07:27,440
So the work that is coming and say, oh yeah, this application is going to run on these two workers, that one is going to run on the other three workers.

74
00:07:27,440 --> 00:07:34,340
So it has a status of the workers, what they're doing, are they online or not, are they busy or not, et cetera.

75
00:07:34,520 --> 00:07:41,040
Every component inside Spark communicates using this Spark RPC protocol that we will detail a little bit later on.

76
00:07:41,040 --> 00:07:46,460
And just to show you what, you know, to demystify this Spark cluster, this is what it looks like.

77
00:07:46,880 --> 00:07:50,600
This is the basic UI of the Spark master on port 8080.

78
00:07:50,600 --> 00:07:52,060
Oh, sorry about that.

79
00:07:52,160 --> 00:07:55,820
So yeah, you can see that we have like, you know, two workers, one which is dead.

80
00:07:55,820 --> 00:07:58,420
We have an application that's running on two cores.

81
00:07:58,420 --> 00:08:02,460
And since each worker has one core, so this application is running on two cores.

82
00:08:02,460 --> 00:08:04,840
It's being distributed over two machines.

83
00:08:05,640 --> 00:08:20,120
But the most important port really is port 7077, because that's the one that we're going to contact in order to send an application, register it, and schedule it to be, you know, executed in parallel on these workers.

84
00:08:20,640 --> 00:08:29,040
And it's going to be these basic hypotheses that we're going to make along this presentation that we have access to that, to this specific ports.

85
00:08:29,360 --> 00:08:33,860
And the third component, and indeed the most component, the most important, sorry, is the driver.

86
00:08:33,860 --> 00:08:38,140
And the driver is really the brain behind the application orchestration.

87
00:08:38,140 --> 00:08:51,420
It's the one that's going to slice up your application and small tasks, contact the cluster manager, receive the worker, and then send, distribute your workload to these worker and, you know, make sure that everything is going smoothly and aggregate the results.

88
00:08:52,500 --> 00:08:56,840
And the driver in a typical case or typical scenario is going to be running on your laptop.

89
00:08:56,840 --> 00:09:11,980
So you enter a network, you have the cluster manager running somewhere, you have the workers running somewhere else, and you're going to boot up the driver, write your application in Java, Scala, Python, or whatever, boot up the driver, and then send it to the cluster manager to be scheduled on the workers,

90
00:09:11,980 --> 00:09:13,520
hopefully to execute some code.

91
00:09:14,080 --> 00:09:18,300
So that's one setup that's kind of common.

92
00:09:18,900 --> 00:09:20,620
Right, so recon.

93
00:09:20,620 --> 00:09:23,540
So let's take this specific scenario that I just listed.

94
00:09:23,540 --> 00:09:28,960
Let's say you end up in a network and you want to find, you want to hunt for some spark in order to exploit it.

95
00:09:29,640 --> 00:09:31,240
How would you go about it?

96
00:09:31,240 --> 00:09:46,280
Well, the first thing you want to do, I guess, is basically if you're in an AWS environment, you can simply use the API, if you have the proper access rights, obviously, to look for every machine that has spark, master, or other keywords in their labels.

97
00:09:46,280 --> 00:09:49,720
Similarly, if you're in a Kube environment, Kubernetes, it's the same thing.

98
00:09:49,720 --> 00:09:53,900
But what if you end up in a more traditional network where you have to nmap the shit out of things?

99
00:09:53,900 --> 00:09:56,020
How can you find that cluster?

100
00:09:56,080 --> 00:09:58,920
Because you see nmap does not really support spark.

101
00:09:58,920 --> 00:10:01,420
So we're in a kind of a pickle here.

102
00:10:01,420 --> 00:10:06,360
So the first thing we want to do is be able to fingerprint spark properly using nmap.

103
00:10:06,360 --> 00:10:10,020
And to do that, I simply whip out a spark cluster in my lab.

104
00:10:10,020 --> 00:10:14,100
I put some Wireshark in the middle, and I analyze the traffic.

105
00:10:14,100 --> 00:10:16,680
And if you do that, this is the blob that you're going to see.

106
00:10:19,240 --> 00:10:25,800
And basically, the way to decompose this is fairly simple, and it's quite repetitive, actually, once you get to know Spark RPC.

107
00:10:25,800 --> 00:10:27,880
So this is the Spark RPC I was talking about.

108
00:10:28,100 --> 00:10:31,860
It always starts with the same thing, the same header.

109
00:10:31,860 --> 00:10:36,640
It's going to be, most of the time, 21 bytes of data followed by the payload, right?

110
00:10:36,640 --> 00:10:46,620
And these 21 bytes of data, a common header when you send Spark RPC commands, is composed of seven bytes of null bytes, followed by two magic bytes.

111
00:10:46,620 --> 00:10:52,840
I call them magic bytes, but basically they depend on the RPC endpoint that you're trying to reach and the type of message that you're sending.

112
00:10:52,900 --> 00:11:00,700
So in this case, I chose the example of verify endpoint, check existence message.

113
00:11:01,260 --> 00:11:04,180
And its magic byte is C305.

114
00:11:04,180 --> 00:11:11,620
If it's, I don't know, if it's a worker sending a heartbeat message to the cluster manager, it's going to be 2B0F or whatever.

115
00:11:11,780 --> 00:11:13,600
But in this case, it's C305.

116
00:11:13,600 --> 00:11:14,820
But anyway, it doesn't matter.

117
00:11:15,300 --> 00:11:18,280
Next, we have eight bytes of data, like random data.

118
00:11:18,280 --> 00:11:19,760
You can put anything you want here.

119
00:11:19,760 --> 00:11:23,100
They'll just be echoed back by the cluster manager.

120
00:11:23,420 --> 00:11:29,560
And finally, we have four bytes of the size of the data that's going to follow, so the payload that will follow.

121
00:11:29,560 --> 00:11:31,120
What is this payload?

122
00:11:31,320 --> 00:11:40,440
Well, you have a bunch of data IP addresses that we don't care about, but the most important thing really, the heart of the payload, if you will, is this serialized object that you can see here.

123
00:11:40,440 --> 00:11:43,800
So it's a serialized JVM object called check existence.

124
00:11:43,800 --> 00:11:46,920
And if you look at the source code, check existence is actually a class.

125
00:11:46,920 --> 00:11:51,840
Well, it's a Scala case class, but just think of it as a regular class, which is immutable.

126
00:11:52,220 --> 00:11:58,420
And it has some interesting other properties in Scala that we simply don't give a fuck about, but yeah, it's a class.

127
00:11:58,960 --> 00:12:02,760
So check existence is a class and it has an attribute name.

128
00:12:02,760 --> 00:12:11,600
So basically the client or driver is sending this serialized class to the cluster manager to see if there's an endpoint bearing the name that it sent.

129
00:12:11,600 --> 00:12:13,300
And the name that it sent is obviously master.

130
00:12:13,300 --> 00:12:20,800
So it wants to know if there is a master RPC endpoint listening on the other end of the communication.

131
00:12:21,080 --> 00:12:29,540
And if that is the case, the cluster manager, well, will respond first with the 21 bytes of data that you can see it echoed back what we said before, right?

132
00:12:29,540 --> 00:12:32,700
And a different magic RPC, magic two bytes.

133
00:12:32,860 --> 00:12:38,720
But anyway, and if there is indeed an endpoint, it sends a Boolean that is set to true.

134
00:12:38,740 --> 00:12:57,740
So obviously I took this special exchange because you can use it and map to fingerprint the cluster manager, but you have to imagine that all Spark RPCs are these exchange of serialized data that are prefixed with some type of header, either a 21 bytes header or a 13 byte header that we will see a bit later on.

135
00:12:57,740 --> 00:13:03,700
But this is the main way that Spark components communicate with each other, just serialized objects.

136
00:13:04,740 --> 00:13:12,540
So once we have this exchange all mapped out, then we can use it and write a Lua script that will replicate the same thing.

137
00:13:12,540 --> 00:13:17,080
Once you do that, obviously, once you deconstruct it like this, it becomes quite obvious to do it.

138
00:13:17,080 --> 00:13:19,660
So I wrote an Nmap script that's on my repo.

139
00:13:20,180 --> 00:13:26,100
Sorry, and I will push it to the real Nmap repo once people have used it enough and debugged it enough.

140
00:13:26,660 --> 00:13:28,980
So yeah, now we are able to identify a Spark cluster.

141
00:13:28,980 --> 00:13:31,100
And you can see they've been true to their promise.

142
00:13:31,100 --> 00:13:33,900
Authentication is indeed disabled by default.

143
00:13:33,900 --> 00:13:35,260
So this is pretty great.

144
00:13:35,980 --> 00:13:37,140
Now comes the interesting part.

145
00:13:37,140 --> 00:13:40,820
Now we know how to locate Spark clusters inside a network.

146
00:13:40,820 --> 00:13:43,700
How can we execute code, right?

147
00:13:43,700 --> 00:13:45,520
Because that's what's really interesting.

148
00:13:45,920 --> 00:14:03,580
And now the standard way of submitting an application on to a Spark cluster is basically you write your application in Java or Scala, what have you, you package it as a JAR file, you boot up the driver, and then you send that JAR file to be executed or parsed,

149
00:14:03,580 --> 00:14:05,800
if you will, by the cluster manager and the workers.

150
00:14:06,440 --> 00:14:15,400
Now, the problem is this, I find writing Java boring, sorry, and Scala makes me want to shoot myself.

151
00:14:15,400 --> 00:14:18,100
So I decided to do it in Python.

152
00:14:18,640 --> 00:14:20,820
And luckily, Spark supports Python.

153
00:14:20,820 --> 00:14:28,160
In fact, they have an official wrapper called PySpark, which actually takes care of all the nasty stuff that I just described.

154
00:14:28,160 --> 00:14:29,840
So it will boot up the driver for you.

155
00:14:29,840 --> 00:14:34,600
It will serialize JVM object into Python object and vice versa and send it.

156
00:14:34,600 --> 00:14:36,620
And it will take care of all the nasty bits for you.

157
00:14:36,620 --> 00:14:39,100
So you just write a simple Python.

158
00:14:40,260 --> 00:14:41,420
And so, yeah, PySpark.

159
00:14:41,420 --> 00:14:44,340
And if you pip install PySpark, it really is just a thin wrapper.

160
00:14:44,340 --> 00:14:46,580
It downloads 200 megabytes of JAR file behind.

161
00:14:46,580 --> 00:14:47,900
So it's hilarious.

162
00:14:47,900 --> 00:14:52,780
Anyway, so from PySpark, we import these two classes, SparkContext and SparkConfiguration.

163
00:14:52,780 --> 00:14:55,900
It's very straightforward, very intuitive, I would say.

164
00:14:55,900 --> 00:14:57,140
So you define the configuration.

165
00:14:57,140 --> 00:14:59,200
We're going to name our application WordCount.

166
00:14:59,200 --> 00:15:02,700
We're going to point to the cluster, port 7077, obviously.

167
00:15:02,820 --> 00:15:07,220
Now you have to define your public IP that you're going to bind on.

168
00:15:07,220 --> 00:15:08,460
We're going to see why later.

169
00:15:08,460 --> 00:15:11,380
But if you forget this line, it will not work.

170
00:15:11,400 --> 00:15:14,620
No word does it say so in the doc, but if you forget this line, it will not work.

171
00:15:14,660 --> 00:15:17,880
So anyway, so you define the SparkContext.

172
00:15:20,300 --> 00:15:26,700
SparkContext is going to be the client that's going to initiate the communication, handle the communication with the cluster manager.

173
00:15:26,700 --> 00:15:32,760
And basically, when the Python interpreter comes to this line, it will boot up the JVM process.

174
00:15:32,760 --> 00:15:36,340
We talked about the driver, and it will initiate communication with the cluster manager.

175
00:15:36,360 --> 00:15:44,380
So quite naively and quite intuitively, I would say, what we want to do next is obviously just send the code that we want to...

176
00:15:45,120 --> 00:15:48,120
we want to be serialized and we want to be executed on the worker.

177
00:15:48,120 --> 00:15:51,660
So from subprocess P open the command ID.

178
00:15:52,540 --> 00:15:55,900
And if you do that, and this is really the first thing that I did.

179
00:15:56,020 --> 00:15:57,980
If you do that, it actually works.

180
00:15:57,980 --> 00:16:01,880
So I got this result back and I was like, well, hang on.

181
00:16:02,160 --> 00:16:04,080
I know that user, that's me.

182
00:16:04,080 --> 00:16:09,300
I have that user defined on my local machine, not on the worker or the cluster machines.

183
00:16:09,300 --> 00:16:10,280
What the fuck happened?

184
00:16:10,280 --> 00:16:13,200
So it turns out I actually just executed the code on my own machine.

185
00:16:13,560 --> 00:16:17,180
So if you look at Wireshark, what happened, because I was curious, what the fuck happened?

186
00:16:17,180 --> 00:16:21,280
I thought this stuff was supposed to serialize and send the command to be executed on the workers.

187
00:16:21,280 --> 00:16:28,920
Well, what happened was first we actually sent this, you know, 21 bytes of data, check existence class, yada, yada, everything's all right.

188
00:16:28,920 --> 00:16:31,100
There is indeed a cluster manager listening.

189
00:16:31,420 --> 00:16:39,000
Then we send the register application class, which contains all the information of the application that we can... that we're going to run.

190
00:16:39,000 --> 00:16:44,660
So you can see that we send, you know, word count plus another other properties, right?

191
00:16:44,660 --> 00:16:48,400
The cluster manager responded with executed added class.

192
00:16:48,580 --> 00:16:51,680
So they gave us actually two workers.

193
00:16:51,680 --> 00:16:52,760
So this is great.

194
00:16:52,760 --> 00:16:53,680
What happened next?

195
00:16:53,680 --> 00:16:59,060
Well, the driver, so our instance of PySpark simply unregistered the application and called off the whole thing.

196
00:16:59,060 --> 00:17:00,340
So what the hell?

197
00:17:00,520 --> 00:17:08,940
Well, to understand what happened actually, remember when I said that Spark contacts, PySpark will simply serialize whatever comes after the Spark contacts line?

198
00:17:08,940 --> 00:17:10,140
Yeah, I lied.

199
00:17:10,660 --> 00:17:15,720
Well, the thing is, in order to understand what happened, we basically need to talk about two concepts.

200
00:17:15,720 --> 00:17:20,680
First of all, Spark APIs, and second, the notion of a DAG and lazy evaluation in Spark.

201
00:17:20,860 --> 00:17:27,480
So Spark APIs, counter-intuitively I would say, Spark APIs are not methods, they're data structures.

202
00:17:28,160 --> 00:17:32,180
So let's say you want to open a file on Python or Java or Scala or whatever.

203
00:17:32,240 --> 00:17:35,740
You point to a file and then you call the method, the open method.

204
00:17:35,740 --> 00:17:43,040
This will only give you an object, a Python object, that refers to a handling memory, whatever, but a single threaded object nonetheless.

205
00:17:43,740 --> 00:17:53,980
If there is no parallelism there, and the proper way to do it in Spark and still get that parallelism is actually to call text file method of the Spark contacts class.

206
00:17:53,980 --> 00:17:56,960
So here SC is an instance of Spark contacts.

207
00:17:56,960 --> 00:18:06,260
So you point to a file, you call the text file method and text file method will grab that file, slice it into fragments and then load those two fragments in memory.

208
00:18:06,920 --> 00:18:11,220
Similarly, if you want to load a list, just don't define a list like that.

209
00:18:11,220 --> 00:18:13,640
This will give you only a single threaded object in memory.

210
00:18:13,640 --> 00:18:21,540
No, you need to call the parallelize method, which will take that list, slice it in half or three or four, whatever Spark seems necessary.

211
00:18:21,740 --> 00:18:27,300
And that way you will end up with two fragments in file on which you can perform work in a parallel fashion.

212
00:18:27,540 --> 00:18:30,080
Now, these fragments are called partitions.

213
00:18:30,140 --> 00:18:33,040
And this is a very frequent term in Spark that will come up a lot.

214
00:18:33,040 --> 00:18:37,180
Partitions are the main unit of work inside Spark.

215
00:18:37,680 --> 00:18:45,340
And these collection of partitions are called resilient distributed datasets, which is really the most fundamental API in Spark.

216
00:18:45,340 --> 00:18:50,300
There are others, but they're ultimately all based on RDDs at the core.

217
00:18:50,500 --> 00:18:58,680
So once we have these partition, then we can apply parallelized work using something called transformations in action.

218
00:18:58,680 --> 00:19:00,000
Now, what's a transformation?

219
00:19:00,340 --> 00:19:01,820
A transformation...

220
00:19:01,820 --> 00:19:05,120
Now, let's say I want to multiply every element on the list by 20, let's say.

221
00:19:05,120 --> 00:19:07,600
A concrete example will actually make it better.

222
00:19:07,780 --> 00:19:13,460
So I define a method called multiply, which takes an element, multiplies it by 20.

223
00:19:13,580 --> 00:19:23,380
And then I just call the map transformation, which will iterate over every element in every partition and yield the same number of elements in the same number of partitions.

224
00:19:23,380 --> 00:19:24,620
So that's a transformation.

225
00:19:24,620 --> 00:19:30,960
There are other types of transformation, which may or may not yield the same number of partitions and elements, by the way.

226
00:19:30,960 --> 00:19:37,060
So yeah, each Spark API basically defines these transformations and actions to work on.

227
00:19:37,500 --> 00:19:45,840
And so what happens is if I write this code, thinking that I just multiplied every element by 20, actually nothing gets sent to the workers just yet.

228
00:19:45,840 --> 00:19:46,700
Nothing happened.

229
00:19:46,700 --> 00:19:47,720
Nothing gets computed.

230
00:19:47,720 --> 00:19:50,340
I will have the same results as I had earlier.

231
00:19:50,340 --> 00:19:51,900
In fact, I would have nothing.

232
00:19:53,520 --> 00:19:54,440
Why?

233
00:19:54,440 --> 00:20:03,460
Because when the driver starts parsing this application, all it does is actually it builds an execution graph.

234
00:20:04,880 --> 00:20:07,640
And this execution graph is called the DAG and Spark terminology.

235
00:20:07,880 --> 00:20:12,420
But basically all it does, it follows all these calls that we made.

236
00:20:12,420 --> 00:20:15,920
So, oh yeah, it knows that it's going to call the paralyzed.

237
00:20:15,920 --> 00:20:18,200
So it knows that it's going to build an RDD.

238
00:20:18,200 --> 00:20:19,260
It doesn't care what's inside.

239
00:20:19,260 --> 00:20:22,780
It just knows that it's going to build an RDD with X number of partitions.

240
00:20:22,860 --> 00:20:24,680
Then, oh yeah, we're going to call a map.

241
00:20:24,680 --> 00:20:26,120
It doesn't care what's inside the map.

242
00:20:26,120 --> 00:20:28,460
It doesn't care that we're going to multiply every element by 20.

243
00:20:28,460 --> 00:20:29,780
That's not the driver's job.

244
00:20:29,780 --> 00:20:34,000
The driver's job is only to follow, to build this execution graph.

245
00:20:34,000 --> 00:20:35,720
So it knows that it's going to apply a map.

246
00:20:35,720 --> 00:20:40,680
So given a set number of partition, it's going to have an output, the same number of partition.

247
00:20:40,680 --> 00:20:42,920
It's going to continue building up this DAG.

248
00:20:42,920 --> 00:20:46,460
Maybe it's going to be a filter map, which is another type of transformation, et cetera.

249
00:20:46,500 --> 00:20:49,280
And it's going to continue building this graph until it hits an action.

250
00:20:49,280 --> 00:20:53,680
An action in Spark is something that's going to force it to do the actual computation.

251
00:20:53,680 --> 00:20:57,220
So think like save file or collect or take sample.

252
00:20:57,220 --> 00:21:02,900
All these methods that you can call in an RDD that will force it to actually have the right value.

253
00:21:02,960 --> 00:21:10,880
And once you do that, once you call a collect or an action in Spark, then, and only then, will the graph be sent to the executors.

254
00:21:11,220 --> 00:21:14,600
The graph, mind you, not the code, simply the graph.

255
00:21:14,980 --> 00:21:16,200
And it goes a little bit like this.

256
00:21:16,200 --> 00:21:19,640
So the driver will send to the workers the graph of execution.

257
00:21:19,640 --> 00:21:24,680
The workers will go through this graph and they go, oh, I need to parallelize, but I don't have that list.

258
00:21:24,680 --> 00:21:25,520
Oh, you know what?

259
00:21:25,540 --> 00:21:26,360
Give me that list.

260
00:21:26,360 --> 00:21:30,220
And the driver will send all these serialized objects, one after the other.

261
00:21:30,380 --> 00:21:31,500
I need to do a map.

262
00:21:31,500 --> 00:21:33,500
Oh, but I don't have the function multiply.

263
00:21:33,740 --> 00:21:39,220
You know, they ask the driver to give, serialize the function, the method, and send it to the workers, et cetera.

264
00:21:39,300 --> 00:21:41,140
And then they're going to apply those methods.

265
00:21:41,140 --> 00:21:46,920
So each worker will only apply these methods, will only follow the DAG on its own partition.

266
00:21:47,500 --> 00:21:51,020
So the worker number one will, you know, loop through the elements of the first partition.

267
00:21:51,020 --> 00:21:56,780
So one, two, three, multiply by 20, 20, 40, 60, send back the result to worker number two, will do the same.

268
00:21:56,780 --> 00:22:01,160
And the driver will aggregate the results and present it to the operator, us.

269
00:22:01,620 --> 00:22:09,960
So when you think about it this way, and you go back to our code earlier with our subprocessor, we're just sitting naked there alone in the wilderness.

270
00:22:10,360 --> 00:22:12,220
Well, of course it didn't make it to the workers.

271
00:22:12,220 --> 00:22:14,640
The workers are only ever aware of the DAG.

272
00:22:14,640 --> 00:22:19,320
And so we need, if we want to execute code, we need to somehow embed it inside the DAG.

273
00:22:19,900 --> 00:22:24,540
So we need to put it inside a transformation, hence the following code.

274
00:22:24,540 --> 00:22:32,120
So here we can see, we define a Lambda function, anonymous function, and inside we put our P open command execution stuff.

275
00:22:32,320 --> 00:22:36,400
And obviously we need to follow it by an action, otherwise nothing gets sent.

276
00:22:36,400 --> 00:22:45,620
And this is the skeleton that you need to follow, or the basic skeleton you need to follow in order to achieve code execution on Spark across multiple machines.

277
00:22:45,760 --> 00:22:55,300
Now, if you don't want to be bothered by creating like, you know, RDDs, handling parallelism and whatnot, I took this skeleton, put it in a fancier way, or something like that.

278
00:22:55,300 --> 00:22:59,780
And I incorporated in a tool called Sparky that will take care of all this stuff for you.

279
00:22:59,780 --> 00:23:11,620
So you just, you know, point to a cluster, point to the IP that you want to target, send the command that you want, and then how many times, on how many workers you want it to execute, and it will do that for you.

280
00:23:11,620 --> 00:23:18,620
It will build the RDDs, build the APIs, call the transformations, the right ones in order to achieve the code, the execution that you want.

281
00:23:19,040 --> 00:23:20,780
And this is what it looks like really.

282
00:23:20,780 --> 00:23:25,400
So again, I'm going to publish it on GitHub for everyone.

283
00:23:25,460 --> 00:23:27,500
So yeah, you'll have a chance to look at the code.

284
00:23:27,500 --> 00:23:34,860
So you can see here, we just point to the cluster, we specify our public IP address, the command that we want to execute, and the number of workers that we want to target.

285
00:23:37,000 --> 00:23:40,740
And you can see here that we have execution on three machines, so great.

286
00:23:40,980 --> 00:23:58,200
I'd also add some, you know, I embedded some scripts to make it easy to run in an AWS environment, to like dump some AWS secrets if you want, or search for files that are sometimes dumped by Spark, you know, work in some work folder.

287
00:23:58,200 --> 00:24:01,800
So you can search for AWS keys or secrets or whatever.

288
00:24:01,800 --> 00:24:05,960
So it automates a lot of the pen tests in Spark.

289
00:24:05,960 --> 00:24:11,080
So yeah, that will get you started.

290
00:24:11,620 --> 00:24:18,260
Next, now in this type of execution that I just showed, there are some problems that are quite annoying.

291
00:24:18,260 --> 00:24:19,640
Let me go through them very quickly.

292
00:24:19,640 --> 00:24:28,300
First of all, the Python version on the driver needs to match the one on the worker, down to the minor version, otherwise it will throw an error.

293
00:24:29,360 --> 00:24:30,380
Now there's a way...

294
00:24:30,380 --> 00:24:35,580
Yeah, so you can see even if you have a worker that's on 3.5 and your driver is on 3.7, you're kind of screwed.

295
00:24:35,580 --> 00:24:37,400
So there's a way around it.

296
00:24:37,740 --> 00:24:41,840
Since the check is simply done by checking the version info, you can just override it.

297
00:24:41,840 --> 00:24:44,360
So you can see here that I put it like 3.5.

298
00:24:44,920 --> 00:24:51,500
And if you do that, it will work, so long as it is not too much of a gap between the two versions.

299
00:24:51,500 --> 00:25:02,220
So it can get away with 3.6 and 3.7 because the serialized objects are similar to some extent, but it cannot get away with the 3.4 and the 3.7 or there's too much of a gap.

300
00:25:02,740 --> 00:25:14,900
And the way around that is simply, well, to override the file, but also to take advantage of the fact that a Spark cluster will use, or PySpark, will use pickle deserializers to deserialize the object.

301
00:25:15,280 --> 00:25:18,160
And in pickle, you can specify a method.

302
00:25:18,160 --> 00:25:26,540
If you define the method reduce of an object, it will get executed first before, in the process of deserialization.

303
00:25:26,540 --> 00:25:36,320
So before it fails, that deserialization process later on, it will have executed the command reduce, which contains the execution command.

304
00:25:36,320 --> 00:25:41,760
And so you simply need to define the object and send the method as part of that object, and it will get you through.

305
00:25:41,760 --> 00:25:49,340
So even though the job submission fails, you still achieve code execution over however many workers you decided.

306
00:25:49,540 --> 00:25:54,060
And I did not incorporate it inside Sparky because it's just a hacky way.

307
00:25:54,060 --> 00:26:02,500
Really the easiest way is simply to align your versions, your version of Python with that of the workers, but oh well, this is just a fun trick to share.

308
00:26:02,640 --> 00:26:07,280
But really the most annoying problem is actually a network problem.

309
00:26:07,340 --> 00:26:22,920
See what happens when we register an application, we send an application, as we contact the driver, as we, sorry, the driver contacts the cluster manager on RPC port 7077, as we just saw, and we'll send the register application class along with the parameters,

310
00:26:22,920 --> 00:26:25,520
the application's name, et cetera.

311
00:26:25,520 --> 00:26:27,560
The cluster manager will receive that application.

312
00:26:27,560 --> 00:26:29,880
Oh yeah, I need three, four workers, whatever.

313
00:26:30,740 --> 00:26:35,260
Gonna fetch the workers, send them to the driver along with some status data.

314
00:26:35,560 --> 00:26:44,700
But then just when you would think that it would be the driver's job to contact the workers, actually, it's the workers who initiate the communication towards the driver.

315
00:26:44,700 --> 00:26:51,940
So they will be the one contacting the driver on a random RPC port called schedule import to say, hey, we're ready, et cetera.

316
00:26:52,440 --> 00:27:01,500
And that's annoying because you don't necessarily have this flow of communication open because the driver is, you know, your computer is usually behind the net or whatever.

317
00:27:01,500 --> 00:27:02,640
So you're kind of screwed.

318
00:27:02,640 --> 00:27:14,840
Not only that, the workers will initiate another communication on another port altogether called the block manager port in order to receive some data blocks, you know, request some serialized objects, broadcast some variables and whatnot.

319
00:27:14,840 --> 00:27:17,320
So that's annoying.

320
00:27:17,320 --> 00:27:27,300
So in Sparky, by default, I basically, I default these settings to some common ports in order to, you know, evade or it works in some settings, but not all.

321
00:27:27,300 --> 00:27:31,220
But really the solution is to use what we call a cluster mode.

322
00:27:31,220 --> 00:27:35,260
And in a cluster mode, you don't run the driver on your laptop.

323
00:27:35,380 --> 00:27:39,400
You simply contact the cluster manager on port 77, 7077, yeah.

324
00:27:39,400 --> 00:27:42,160
And then you tell it, oh, here's the application to execute.

325
00:27:42,160 --> 00:27:46,660
And please also set up the driver on your own cluster.

326
00:27:47,860 --> 00:27:49,540
And that works pretty well.

327
00:27:49,540 --> 00:27:52,740
So in this case, we only need access to port 7077.

328
00:27:52,940 --> 00:27:56,320
The downside is that you cannot use PySpark.

329
00:27:56,320 --> 00:27:57,580
You cannot use Python.

330
00:27:57,580 --> 00:28:01,860
So I had to redo that skeleton I showed you earlier in Python.

331
00:28:01,860 --> 00:28:13,620
I had to do it in Scala to build a jar file that will parallelize, that will execute whatever command you give it across however many servers that you want.

332
00:28:13,620 --> 00:28:18,500
So yeah, the code is again on GitHub, but you don't need to touch this one.

333
00:28:18,500 --> 00:28:20,400
It's this compiled jar file.

334
00:28:21,240 --> 00:28:23,580
So you can just use it as it is.

335
00:28:23,760 --> 00:28:25,200
It's, yeah.

336
00:28:25,460 --> 00:28:27,480
So here's a quick example.

337
00:28:27,480 --> 00:28:32,460
So you can see we host this jar file that I just showed you on a server that is reachable by the workers.

338
00:28:32,460 --> 00:28:37,420
And we simply use Sparky again to build that command to register this in cluster mode.

339
00:28:37,420 --> 00:28:41,560
So we just point to this cluster, to this jar file command we want to execute.

340
00:28:41,560 --> 00:28:47,960
And then we don't need any network condition except for that 7077 to reach the cluster.

341
00:28:47,960 --> 00:28:49,660
And it works pretty well.

342
00:28:49,840 --> 00:28:51,660
It will even auto delete itself.

343
00:28:51,660 --> 00:28:53,180
So you don't have to worry about that.

344
00:28:53,180 --> 00:28:56,320
So this gets us through the network thing.

345
00:28:56,780 --> 00:29:06,280
So these are the main ways to execute code on Spark that I found on a Spark cluster that has no authentication, which is enabled by default.

346
00:29:06,420 --> 00:29:10,040
So anyway, let's go explore some other facets of Spark.

347
00:29:10,260 --> 00:29:14,660
And there's this interesting interface called, well, it's the REST API.

348
00:29:14,660 --> 00:29:20,840
And it is interesting because when you go through the documentation, you can see that it's available in port 6066.

349
00:29:21,160 --> 00:29:25,960
And when you look at the documentation, it's really a read-only boring API.

350
00:29:25,960 --> 00:29:27,480
I was like, yeah, whatever.

351
00:29:27,480 --> 00:29:37,280
And I remember a few days later, I was digging through the source code, like really lost inside the source code amongst weird scholarship functions, options and whatnot.

352
00:29:37,280 --> 00:29:40,080
And I remember I saw this keyword REST.

353
00:29:40,080 --> 00:29:41,660
I'm like, oh my God, I know what that is.

354
00:29:41,660 --> 00:29:43,460
And I grabbed it and I followed it.

355
00:29:43,480 --> 00:29:48,920
And I came across this piece of code that says something like, submit requests, create.

356
00:29:48,920 --> 00:29:50,600
I was like, I like to submit stuff.

357
00:29:50,600 --> 00:29:51,760
I like to create stuff.

358
00:29:51,760 --> 00:29:52,900
So this thing do.

359
00:29:53,980 --> 00:29:57,100
And you follow this, create submission requests.

360
00:29:57,100 --> 00:30:00,000
And I was like , let's talk about app resource, main class.

361
00:30:00,000 --> 00:30:01,300
I'm like, well, hang on.

362
00:30:01,300 --> 00:30:05,840
App resources references a jar usually in Spark code.

363
00:30:05,840 --> 00:30:08,500
So I thought this was a read-only API.

364
00:30:08,500 --> 00:30:09,620
What the hell is going on?

365
00:30:09,620 --> 00:30:12,560
So I followed this, create submission request.

366
00:30:12,560 --> 00:30:13,580
And I came across this.

367
00:30:13,580 --> 00:30:16,080
Yeah, app resources, indeed an application jar.

368
00:30:16,080 --> 00:30:26,560
So basically the REST API, even though the documentation, it mentions it's only a read-only API, actually in the code, it accepts a jar file that you can send it, which is kind of amazing.

369
00:30:27,040 --> 00:30:30,440
All you have to do is simply build a JSON file.

370
00:30:30,440 --> 00:30:31,820
It accepts a JSON input.

371
00:30:31,820 --> 00:30:34,000
So you build a JSON file with the right properties.

372
00:30:34,000 --> 00:30:37,480
So the app resource, main class, Spark properties and so on.

373
00:30:37,480 --> 00:30:40,940
And then you can send your jar file and then chief code execution.

374
00:30:40,940 --> 00:30:44,240
And we can take that jar file we used earlier.

375
00:30:44,240 --> 00:30:46,520
So simple app jar, send it a command arc.

376
00:30:46,520 --> 00:30:49,840
So this is, it just accepts commands to execute in base 64.

377
00:30:49,840 --> 00:30:57,760
So this is really, I think a, who am I, or some shit like, no, not who am I, but touch some file in some folder in base 64.

378
00:30:57,760 --> 00:30:58,580
It's very simple.

379
00:30:59,080 --> 00:31:05,620
And yeah, but there's different properties and you can just curl this payload and it will execute codes as simple as that.

380
00:31:06,480 --> 00:31:09,000
And this is just to show you what it looks like.

381
00:31:09,560 --> 00:31:15,520
But yeah, but again, if you use this method, the driver, you don't need it to run on your machine.

382
00:31:15,520 --> 00:31:17,040
It's going to run in the cluster.

383
00:31:17,040 --> 00:31:18,320
So it's going to run in cluster mode.

384
00:31:18,320 --> 00:31:24,440
And one thing I will mention, if you decide to use the REST API or cluster mode in general, you can send whatever jar file you want.

385
00:31:24,440 --> 00:31:29,200
You don't have to use my specific jar file, but if you, you will get execution on one worker.

386
00:31:29,200 --> 00:31:34,760
But if you want to execute on 200, 300, all the workers inside the cluster, then you need to use Spark API.

387
00:31:34,760 --> 00:31:44,840
So you need to use some version of the Spark, like, you know, Spark APIs I just showed in the Scala code earlier, or Java code earlier.

388
00:31:44,840 --> 00:31:50,220
So anyway, so this was fun and I thought, well, I wonder if anybody blogged about it.

389
00:31:50,220 --> 00:31:52,780
And I Googled create submission request.

390
00:31:52,780 --> 00:31:57,580
And I came across the first result on Google was this script and it was already published.

391
00:31:57,580 --> 00:31:59,640
And I was like, fuck, so close.

392
00:31:59,640 --> 00:32:04,280
Not only that, there was actually an exploit that was already, you know, published.

393
00:32:04,280 --> 00:32:08,660
It was actually a Metasploit model that was published by these fine gentlemen.

394
00:32:08,660 --> 00:32:09,920
I was like, damn it.

395
00:32:10,180 --> 00:32:17,800
Just, you know, sometimes you keep your head in the wheel and you forget to do basic research, but it was just, you know, months away.

396
00:32:17,960 --> 00:32:21,460
Anyway, let's get to the real stuff, right?

397
00:32:22,840 --> 00:32:24,040
So authentication.

398
00:32:24,280 --> 00:32:29,180
Let's say you come across a cluster that has all the right switches on, right?

399
00:32:29,180 --> 00:32:30,800
So it's enabled authentication.

400
00:32:30,800 --> 00:32:32,880
And the way to do that in Spark is fairly easy.

401
00:32:32,880 --> 00:32:35,200
You just enable authenticate to true.

402
00:32:35,200 --> 00:32:37,500
Then you define a secret, you can see here.

403
00:32:38,040 --> 00:32:42,100
And yeah, all the components you do, you need to do that on every component inside the cluster.

404
00:32:42,100 --> 00:32:46,320
So every worker and every cluster manager and the driver.

405
00:32:46,360 --> 00:32:48,820
And that way they all have a shared secret to communicate with.

406
00:32:48,820 --> 00:33:00,820
And if you have an authenticated cluster and you try to communicate with it, you will get this delightful error that says, you know, you send this check existence class, and then you get this illegal state exception, ACSL message.

407
00:33:00,820 --> 00:33:09,320
Now, ACSL is an authentication protocol, challenge response protocol, based on a shared secret, as you might've guessed.

408
00:33:09,920 --> 00:33:20,240
But yeah, basically the server sends a nonce to the client and the client hashes that with the nonce, hashes the secrets with the nonce, and then sends the response plus other parameters that we will see, but fairly simple.

409
00:33:20,240 --> 00:33:30,020
But the funny thing is that even though the driver, you configure the driver to use the secret and you tell it to authenticate, it will first attempt an unauthenticated session.

410
00:33:30,020 --> 00:33:38,260
So it will first, you know, try check existence, get slapped with this error, and then it will try an authenticated message.

411
00:33:38,260 --> 00:33:39,020
So it's kind of weird.

412
00:33:39,020 --> 00:33:42,080
So anyway, what does it look like, an authenticated session?

413
00:33:42,080 --> 00:33:58,060
Well, first the driver will send 21 bytes of data, the header, the usual header, with the magic bytes, this time set to 2B03, as opposed to C05 earlier, followed by, you know, eight random bytes that we don't care about, the size of the payload, which is definitely wrong in this case,

414
00:33:58,060 --> 00:34:02,140
because we're not talking, it's not 176 bytes.

415
00:34:02,140 --> 00:34:12,480
But anyway, the most important thing in this authentication sequence is that it sends the Spark ACSL user, which is going to be the user used inside the challenge calculation.

416
00:34:12,920 --> 00:34:22,720
So anyway, Cluster Manager responds with 21 bytes data as usual, the header, and then it follows by the parameters to, ACSL parameters to perform the challenge computation.

417
00:34:22,720 --> 00:34:29,680
So you have the nonce, random value, realm, QOP, which is the, are we doing authentication or encryption or both?

418
00:34:29,680 --> 00:34:34,180
These delightful algorithms to use in case we're doing ACSL encryption.

419
00:34:34,180 --> 00:34:39,800
Now, just to reassure you, Spark does not use ACSL encryption by default.

420
00:34:39,980 --> 00:34:43,820
It uses AES, but it's definitely possible to use it as a fallback.

421
00:34:43,820 --> 00:34:48,420
And the algorithm, and this one is indeed used to calculate the hash.

422
00:34:48,420 --> 00:34:51,040
And the calculation of the hash, we're not going to do it, honestly.

423
00:34:52,100 --> 00:34:59,380
It's detailed in RFC 2831, but you take a bunch of data, you hash it in MD5, you take the output, you combine it with the nonce and other stuff.

424
00:34:59,380 --> 00:35:07,720
And then you hashed a bunch, you hash other types of data, and then you combine everything, which gives you the response to the challenge.

425
00:35:07,760 --> 00:35:11,320
Again, if you want to know more about it, Google it, RFC 2831.

426
00:35:11,320 --> 00:35:16,260
So, the driver sends the response to the cluster manager.

427
00:35:16,840 --> 00:35:19,500
If everything checks out, the cluster manager acts.

428
00:35:20,100 --> 00:35:26,320
Right, and then it can follow the classic flow of communication.

429
00:35:26,320 --> 00:35:35,140
So basically, check existence class, register application, register with application, executors added, yada, yada, yada.

430
00:35:35,560 --> 00:35:51,640
Now, here's what bothered me, or what bothered me, what ticked me a little bit, is that this authentication step was only ever done when there was an RPC endpoint that was prefixed, an RPC-ish communication that was prefixed with those 21 bytes of data.

431
00:35:51,640 --> 00:35:59,960
But it turns out, there's another header that is sometimes sent before payloads, that prefixes payloads.

432
00:35:59,960 --> 00:36:02,520
And this header is only 13 bytes long.

433
00:36:02,520 --> 00:36:07,500
And it's usually sent inside the middle, inside the TCP stream.

434
00:36:07,500 --> 00:36:09,400
So it's inside the same TCP session.

435
00:36:09,400 --> 00:36:11,220
So I did not have much hope.

436
00:36:11,220 --> 00:36:15,260
I was like, well, maybe the cluster actually remembers that it's in an authenticated session.

437
00:36:15,260 --> 00:36:17,120
So that's why, et cetera.

438
00:36:17,140 --> 00:36:19,580
But nevertheless, I thought, well, you know what?

439
00:36:19,580 --> 00:36:21,140
Let's see what it looks like.

440
00:36:21,140 --> 00:36:24,440
Let's see if we start the communication from here.

441
00:36:24,440 --> 00:36:29,440
So we don't send this check existence header, which prompts the authentication message.

442
00:36:29,440 --> 00:36:37,320
Let's just start communication from here, by sending these 13 bytes of data, followed by the register application, because that's what we really care about, right?

443
00:36:37,320 --> 00:36:38,960
We want to submit an application.

444
00:36:39,560 --> 00:36:40,880
That's what we really want.

445
00:36:41,180 --> 00:36:47,080
And so if you begin communication at this stage, you first send 13 bytes of data.

446
00:36:47,580 --> 00:36:52,720
It looks awfully similar to the 21 bytes header, but basically, oh, sorry.

447
00:36:52,840 --> 00:36:57,540
You have four bytes, four null bytes, followed by the size of the payload plus the header.

448
00:36:57,540 --> 00:37:04,900
So the size plus 13, a magic byte nine, which didn't change for some reason, and four bytes of data, which are the actual size of the payload.

449
00:37:04,900 --> 00:37:10,420
And then you send the serialized register application with all its attributes, you know, application's name, yeah, yeah.

450
00:37:10,900 --> 00:37:21,800
And I sent this, and I was expecting a massive error, like, you know, those JVM hundred line errors, but instead I got a header followed by register application class.

451
00:37:21,800 --> 00:37:25,920
And I was like, is it me, or is it just possible to bypass authentication?

452
00:37:25,920 --> 00:37:28,200
I was like, hallelujah, it's amazing.

453
00:37:28,940 --> 00:37:33,860
Now, bypassing authentication is fine, but what can we do with it?

454
00:37:34,420 --> 00:37:41,300
Because I was not going to recreate the 80 next messages that were following along, it's not possible.

455
00:37:41,520 --> 00:37:43,980
So what can we do with just one message?

456
00:37:43,980 --> 00:38:00,940
Well, if you look back at Spark, how it works, the register application class is the one that prompts all that workload on the cluster manager that's gonna contact the workers and say, hey, I've got an application for you, you better get ready, spawn those executors,

457
00:38:00,940 --> 00:38:02,700
and get ready to take the work.

458
00:38:03,300 --> 00:38:10,120
So I went ahead to the workers and I replayed my message, hoping to see some process spawn or whatever, but I didn't see anything.

459
00:38:10,540 --> 00:38:21,200
And the reason for it, I figured out a couple of days later, was because when you send this payload, serialized register application, when you send it, the cluster manager receives it.

460
00:38:22,320 --> 00:38:25,620
I immediately shut down the connection, the program terminated.

461
00:38:25,620 --> 00:38:32,900
So the cluster manager did not have enough time to actually contact the worker, spawn the process, and let it do its thing.

462
00:38:33,160 --> 00:38:36,260
Now, so I immediately shut down the connection.

463
00:38:36,260 --> 00:38:40,220
So the way around it, in order to give time, you simply need to do a sleep.

464
00:38:40,220 --> 00:38:41,420
As simple as that.

465
00:38:41,420 --> 00:38:44,400
You send the register application, you sleep for five seconds.

466
00:38:44,420 --> 00:38:50,540
That way you give enough time for the cluster manager to contact the worker, spawn the executor, and do its work.

467
00:38:50,540 --> 00:39:00,160
And once you do that, once you do that sleep, I saw, like I put a watcher on the worker, monitoring for processes, and I saw this process getting started.

468
00:39:00,280 --> 00:39:09,700
And the magnificent thing about it was that it took as parameters, something that I was, I was defining inside that serialized register application.

469
00:39:09,700 --> 00:39:11,920
You know, the block manager report set to 8443.

470
00:39:11,920 --> 00:39:18,220
And I knew it was me because I'm the one that said it in order to have communication come from the worker to the driver.

471
00:39:18,580 --> 00:39:22,440
So I thought, okay, authentication bypass is a sure thing.

472
00:39:22,860 --> 00:39:24,960
Remote code execution is a definite maybe.

473
00:39:24,960 --> 00:39:28,560
Because you can influence parameters that are being used to spawn a process.

474
00:39:28,560 --> 00:39:30,380
So maybe there's some injection there.

475
00:39:30,380 --> 00:39:35,020
So I turned to this register application class and saw like, okay, what's inside?

476
00:39:35,020 --> 00:39:38,700
What can we define as parameters as options that we can take an advantage of?

477
00:39:38,700 --> 00:39:44,360
So register application, again, a case class, which extends the deploy message, which implements the trade serializable.

478
00:39:44,460 --> 00:39:45,720
Don't care about that.

479
00:39:45,720 --> 00:39:47,560
Application description, what does it do?

480
00:39:47,560 --> 00:39:53,960
Okay, so if you're like me, you're going to ignore all this shit and go straight to command, right?

481
00:39:53,960 --> 00:40:00,060
Because the command class, so command case class, it has the main class, which is the class of the executor to be executed.

482
00:40:00,060 --> 00:40:01,580
We cannot overwrite that.

483
00:40:01,620 --> 00:40:04,280
You have arguments, environments, a bunch of parameters.

484
00:40:04,320 --> 00:40:13,180
Now, if you just want to inject like, you know, a semicolon or an and or a pipe, it will not work because these parameters are sanitized.

485
00:40:13,600 --> 00:40:17,700
So we need a clever, more clever way of hijacking the execution.

486
00:40:17,700 --> 00:40:22,540
And the way to do it, I found the easiest way to do it at least, is using these Java options.

487
00:40:22,540 --> 00:40:28,300
Now, if you're familiar with the JVM, you know that you can actually specify or control the behavior of that JVM.

488
00:40:28,300 --> 00:40:35,720
Like, I don't know which garbage collector to use, how should strings be compressed, the amount of heap memory, et cetera.

489
00:40:35,920 --> 00:40:43,180
Well, it turns out if you dig into the hundreds and hundreds of Java options, there is one actually that allow code execution.

490
00:40:43,280 --> 00:40:45,660
Also, yeah, this is an example of a JVM option.

491
00:40:45,840 --> 00:40:50,100
So there is one that allows code execution and it's called an un-out-of-memory error.

492
00:40:50,100 --> 00:40:52,760
And you just specify a command and it will execute it.

493
00:40:52,760 --> 00:40:56,960
The catch is that it only executes it if there's an un-out-of-memory error.

494
00:40:56,960 --> 00:40:58,800
So like its name said.

495
00:40:59,120 --> 00:41:07,660
So if the executor or process fails to allocate memory, heap memory, then it will throw this error and then it will execute code.

496
00:41:07,660 --> 00:41:10,520
So how to make the executor trigger this error?

497
00:41:10,520 --> 00:41:19,300
Well, simple way is simply to add another JVM option, xmx, which will set the maximum amount of memory to a ridiculous amount, like one megabyte or two megabytes.

498
00:41:19,300 --> 00:41:23,920
And once you combine these two properties, then you have your code execution.

499
00:41:24,580 --> 00:41:26,720
And so ideally, we'd like to do something like this.

500
00:41:26,720 --> 00:41:32,340
Obviously, we cannot do it inside the code because while the driver will attempt to authenticate, which is no good.

501
00:41:32,340 --> 00:41:34,940
And anyway, you cannot set the xmx inside the driver.

502
00:41:34,940 --> 00:41:45,940
So we really need to forge by hand, basically, that serialized object, embed these options, have a properly serialized object and send it to the cluster manager.

503
00:41:45,940 --> 00:41:49,820
And again, easy way to do it using Sparky.

504
00:41:51,780 --> 00:41:53,300
So I just point...

505
00:41:53,300 --> 00:41:56,420
So this is just a quick example to show you how to exploit this phone.

506
00:41:57,220 --> 00:42:00,420
So this is the reversal that we want to execute, right?

507
00:42:00,420 --> 00:42:07,660
So I just point to Sparky, I want to execute this code, but as you can see, it requires, the cluster requires authentication.

508
00:42:09,820 --> 00:42:17,320
But if I add dash B, then it will take care, it will take advantage of this phone and you can see it bypassed authentication, which has executed code directly.

509
00:42:17,320 --> 00:42:21,020
And it's gonna execute it on as many workers as requested.

510
00:42:21,700 --> 00:42:25,720
So yeah, we just bypassed authentication, which has executed code on workers.

511
00:42:26,740 --> 00:42:27,840
So great.

512
00:42:28,920 --> 00:42:36,980
So yeah, so now that we understood like the hacker intuitive way of finding this phone, let's go through the code, because I think it's interesting.

513
00:42:37,040 --> 00:42:45,520
If you see, like, I made the distinction between the 21 bytes, the, you know, payload that was prefixed with 21 bytes of data or prefixed with 13 bytes of data.

514
00:42:45,520 --> 00:42:49,000
It turns out there's actually a difference in the dispatcher.

515
00:42:49,040 --> 00:42:52,960
So the register application is handled by the process one-way message.

516
00:42:53,020 --> 00:42:56,580
And the process one-way message called the receive method of the RPC handler, right?

517
00:42:56,580 --> 00:43:02,540
So it delegates to the receive, and you can see there at no point there is any authentication going on here.

518
00:43:02,540 --> 00:43:04,900
The delegate receive will actually perform the work.

519
00:43:04,900 --> 00:43:13,680
If you compare that with the other receive method, which actually, you know, asks for authentication, you can see that it's quite beefy and different.

520
00:43:13,680 --> 00:43:27,100
So the correction, basically, the patch was to protect that same receive method and another one used in streaming, which was affected also, by the same type, the same types of checks.

521
00:43:27,220 --> 00:43:30,520
And this was labeled CB20209480.

522
00:43:30,520 --> 00:43:36,100
I disclosed them on 24th of December, if I'm not mistaken, which was very horrible for me.

523
00:43:36,100 --> 00:43:39,640
But anyway, and yeah, it was fixed by the Apache Spark team.

524
00:43:39,640 --> 00:43:41,780
So thank you very much for the work that you guys have done.

525
00:43:41,780 --> 00:43:42,720
It's amazing.

526
00:43:43,380 --> 00:43:48,180
Yeah, so in summary, what I wanna say is Spark is awesome.

527
00:43:48,280 --> 00:43:53,180
I'm gonna say too bad security is not taken seriously, and I'm not talking about the folks who manage security there.

528
00:43:53,480 --> 00:43:57,300
I'm just referring to the fact that security is off by default.

529
00:43:57,300 --> 00:44:04,660
I don't think that should be something that we should have in 2020, especially for something that's a framework that does, you know, data computation.

530
00:44:04,940 --> 00:44:06,740
And we all know how data is valuable.

531
00:44:06,740 --> 00:44:10,960
So I say it in reference to that.

532
00:44:11,640 --> 00:44:13,660
Anyway, I hope that will change.

533
00:44:13,860 --> 00:44:20,820
And finally, yeah, we only covered Spark standalone where the cluster manager is actually the one doing the work.

534
00:44:20,820 --> 00:44:28,040
But you can have other setups where it's actually Yarn that's doing the work, or I think even a Kubernetes API server in Spark 3.

535
00:44:28,040 --> 00:44:30,420
So yeah, there's a lot of ground to cover.

536
00:44:30,420 --> 00:44:32,680
So if you wanna dig into it, go ahead.

537
00:44:33,560 --> 00:44:41,860
Yeah, the code to Sparky, if you wanna check out these, you know, Spark APIs, the serialized object that's being forced in order to bypass the vuln, all that stuff, it's there.

538
00:44:41,860 --> 00:44:49,500
I hope you got enough of this talk to actually dig into it because that's really the only way to figure out how stuff works.

539
00:44:49,520 --> 00:44:52,460
So yeah, your contribution are most welcome.

540
00:44:52,460 --> 00:44:55,960
And if you wanna hit me up on Twitter to talk about it more, please do not hesitate.

541
00:44:55,960 --> 00:44:59,500
And looking forward to talk to you guys and to talk to you folks in the QA.

542
00:44:59,500 --> 00:45:00,520
Thank you very much.

543
00:45:00,520 --> 00:45:01,200
Bye-bye.